diff --git a/lab-04-data-pipeline/.gitignore b/lab-04-data-pipeline/.gitignore new file mode 100644 index 0000000..850a138 --- /dev/null +++ b/lab-04-data-pipeline/.gitignore @@ -0,0 +1,19 @@ +# Archivos generados por el pipeline — no commitear +data/processed/ + +# Incluir explícitamente los datos de ejemplo GTFS +# (el .gitignore raíz excluye *.txt en data/) +!data/sample_gtfs/*.txt + +# Python +__pycache__/ +*.pyc +*.pyo +.pytest_cache/ + +# Jupyter checkpoints +.ipynb_checkpoints/ + +# Entornos virtuales +.venv/ +venv/ diff --git a/lab-04-data-pipeline/Dockerfile b/lab-04-data-pipeline/Dockerfile new file mode 100644 index 0000000..617e546 --- /dev/null +++ b/lab-04-data-pipeline/Dockerfile @@ -0,0 +1,21 @@ +FROM python:3.12-slim + +COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv + +ENV UV_SYSTEM_PYTHON=1 \ + PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 + +WORKDIR /app + +RUN uv pip install --system \ + "prefect>=3.0" \ + "polars>=1.0" \ + "pyarrow>=16.0" \ + "psycopg2-binary>=2.9" \ + "python-dotenv>=1.0" \ + "pytest>=8.0" \ + "pytest-asyncio>=0.23" \ + "jupyterlab>=4.0" + +COPY . . diff --git a/lab-04-data-pipeline/README.md b/lab-04-data-pipeline/README.md new file mode 100644 index 0000000..16e6df1 --- /dev/null +++ b/lab-04-data-pipeline/README.md @@ -0,0 +1,183 @@ +# Lab 04 — Data Pipeline: Prefect · TimescaleDB · Polars · GTFS + +Pipeline ETL completo para datos de transporte público en formato GTFS. +Ingesta, transforma y almacena datos de paradas, rutas y eventos de vehículos +usando orquestación con Prefect y series temporales con TimescaleDB. + + +--- + +## Stack + +| Tecnología | Rol | +|-----------------|-------------------------------------------------------------| +| **Prefect 3** | Orquestación de flows: dependencias, reintentos, UI | +| **TimescaleDB** | PostgreSQL + hypertables para datos de series temporales | +| **Polars** | Transformaciones de DataFrames, lectura de GTFS y Parquet | +| **GTFS** | Formato estándar de datos de transporte público | +| **Jupyter** | Análisis exploratorio interactivo | + +--- + +## Conceptos + +### Prefect vs Celery +Celery es una cola de tareas **reactiva**: ejecuta una tarea cuando llega un mensaje. +Prefect es un **orquestador de workflows**: define grafos de tareas con dependencias, +reintentos con backoff, observabilidad centralizada y UI para monitorear runs. + +``` +Celery → "cuando llegue un mensaje MQTT, procesar esta posición" +Prefect → "cada hora: descargar el feed GTFS → validar → transformar → cargar a la BD" +``` + +### TimescaleDB vs PostgreSQL +TimescaleDB **es** PostgreSQL: mismo driver (`psycopg2`), mismas queries SQL. +Agrega dos primitivos clave: + +- **Hypertable**: tabla particionada automáticamente por tiempo. Una query + `WHERE timestamp > NOW() - INTERVAL '1 hour'` solo toca la partición relevante. +- **`time_bucket()`**: agrupación por intervalos de tiempo, equivalente al + `DATE_TRUNC` de SQL estándar pero más flexible. + +```sql +-- Velocidad promedio por vehículo en ventanas de 5 minutos +SELECT time_bucket('5 minutes', recorded_at) AS bucket, + vehicle_id, + AVG(speed_kmh) AS avg_speed +FROM vehicle_events +WHERE recorded_at > NOW() - INTERVAL '1 hour' +GROUP BY bucket, vehicle_id +ORDER BY bucket; +``` + +### Polars vs Pandas +Polars usa Apache Arrow internamente. Sus ventajas en un pipeline ETL: + +| Característica | Polars | Pandas | +|-------------------|-------------------------------|-------------------------| +| Motor | Rust + Apache Arrow | NumPy / Python | +| Evaluación | Lazy (optimiza el plan) | Eager (ejecuta al vuelo)| +| Lectura Parquet | Cero copias | Copia en memoria | +| Velocidad típica | 5–20× más rápido | Baseline | + +### GTFS (General Transit Feed Specification) +Formato estándar para datos de transporte público (Google Maps, OSM, SIMOVI). +Consiste en archivos `.txt` (CSV) con nombres fijos: + +| Archivo | Contenido | +|------------------|--------------------------------------------------| +| `stops.txt` | Paradas: id, nombre, latitud, longitud | +| `routes.txt` | Rutas: id, nombre corto, nombre largo, tipo | +| `trips.txt` | Viajes: ruta, servicio, destino | +| `stop_times.txt` | Horarios: llegada y salida por parada | +| `calendar.txt` | Calendario de servicios: días activos | + +--- + +## Estructura + +``` +lab-04-data-pipeline/ +├── Dockerfile +├── docker-compose.yml +├── pyproject.toml +├── data/ +│ └── sample_gtfs/ # Feed GTFS de ejemplo (buses San José, CR) +│ ├── stops.txt +│ ├── routes.txt +│ ├── trips.txt +│ ├── stop_times.txt +│ └── calendar.txt +├── models/ +│ └── timescale_schema.sql # DDL: tablas normales + hypertables +├── flows/ +│ ├── ingest_gtfs.py # Flow: GTFS → validar → cargar a TimescaleDB +│ ├── transform.py # Flow: Polars → estadísticas → Parquet +│ └── analyze.py # Flow: queries con time_bucket() +├── notebooks/ +│ ├── explore_data.ipynb # Análisis descriptivo: GTFS + series temporales +│ └── ml_demand.ipynb # Predicción de demanda con rolling mean (Polars) +└── tests/ + ├── conftest.py # Fixtures: schema, aislamiento de tablas + └── test_flows.py # 17 tests de integración (TimescaleDB real) +``` + +--- + +## Servicios Docker + +| Servicio | Puerto host | Descripción | +|-------------------|-------------|-------------------------------------------| +| `db` | 5433 | TimescaleDB (pg 16) — 5433 evita colisión con otros labs | +| `prefect-server` | 4200 | UI y API de Prefect | +| `prefect-worker` | — | Ejecuta los flows en un work pool | +| `app` | 8888 | Flows ad-hoc + JupyterLab | + +--- + +## Inicio rápido + +```bash +# 1. Levantar servicios +docker compose up -d + +# 2. Correr el pipeline completo (el schema se aplica automáticamente en ingest) +docker compose exec app python flows/ingest_gtfs.py +docker compose exec app python flows/transform.py +docker compose exec app python flows/analyze.py + +# 3. Tests +docker compose exec app pytest tests/ -v + +# 4. UI de Prefect — ver runs, tasks y logs +# http://localhost:4200 +``` + +--- + +## Notebooks de análisis + +Los notebooks **no son parte del pipeline productivo** — son herramientas de +exploración interactiva que se usan después de que el pipeline ya cargó los datos. + +| Notebook | Contenido | +|---------------------|---------------------------------------------------------------| +| `explore_data.ipynb`| Resumen del feed GTFS, estadísticas por ruta, series temporales con `time_bucket()` | +| `ml_demand.ipynb` | Feature engineering con Polars (lags, rolling mean), predicción baseline de abordajes | + +Para abrirlos, lanzar JupyterLab desde el contenedor `app`: + +```bash +docker compose exec app jupyter lab \ + --ip=0.0.0.0 --port=8888 --no-browser --allow-root \ + --notebook-dir=/app/notebooks +``` + +Luego abrir la URL con token que aparece en la consola (ej: `http://127.0.0.1:8888/lab?token=...`). + +> **Nota**: ejecutar los tres flows antes de abrir los notebooks. +> Las celdas de series temporales (secciones 5–7 de `explore_data`) requieren +> datos en `vehicle_events` y `stop_ridership`, que genera `flows/analyze.py`. + +--- + +## Hypertables creadas + +| Tabla | Tipo | Columna de tiempo | Descripción | +|--------------------|-------------|--------------------|------------------------------------| +| `stops` | Normal | — | Paradas del feed GTFS | +| `routes` | Normal | — | Rutas del feed GTFS | +| `trips` | Normal | — | Viajes del feed GTFS | +| `vehicle_events` | Hypertable | `recorded_at` | Posición + velocidad por vehículo | +| `stop_ridership` | Hypertable | `recorded_at` | Pasajeros por parada por intervalo | + +--- + +## Qué demuestra este laboratorio + +- **Orquestación de pipelines** con Prefect 3: flows, tasks, reintentos automáticos y UI de observabilidad +- **Series temporales** con TimescaleDB: creación de hypertables, particionado automático por tiempo y uso de `time_bucket()` para agregaciones eficientes +- **Transformaciones de alto rendimiento** con Polars: lectura de archivos GTFS (CSV), limpieza, joins y exportación a Parquet +- **Modelado de datos de transporte**: ingesta de un feed GTFS completo (paradas, rutas, viajes, horarios) en una base de datos relacional +- **Pipeline ETL reproducible**: cada etapa (ingest → transform → analyze) es un flow independiente que puede ejecutarse y monitorearse por separado \ No newline at end of file diff --git a/lab-04-data-pipeline/data/sample_gtfs/.gitkeep b/lab-04-data-pipeline/data/sample_gtfs/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/lab-04-data-pipeline/data/sample_gtfs/calendar.txt b/lab-04-data-pipeline/data/sample_gtfs/calendar.txt new file mode 100644 index 0000000..f1631ac --- /dev/null +++ b/lab-04-data-pipeline/data/sample_gtfs/calendar.txt @@ -0,0 +1,4 @@ +service_id,monday,tuesday,wednesday,thursday,friday,saturday,sunday,start_date,end_date +WEEKDAY,1,1,1,1,1,0,0,20260101,20261231 +WEEKEND,0,0,0,0,0,1,1,20260101,20261231 +DAILY,1,1,1,1,1,1,1,20260101,20261231 diff --git a/lab-04-data-pipeline/data/sample_gtfs/routes.txt b/lab-04-data-pipeline/data/sample_gtfs/routes.txt new file mode 100644 index 0000000..1726c51 --- /dev/null +++ b/lab-04-data-pipeline/data/sample_gtfs/routes.txt @@ -0,0 +1,6 @@ +route_id,route_short_name,route_long_name,route_type,route_color,route_text_color +R01,102,San José - Escazú vía La Sabana,3,E8000A,FFFFFF +R02,328,San José - San Pedro vía UCR,3,0058A8,FFFFFF +R03,401,San José - Desamparados,3,00A651,FFFFFF +R04,310,San José - Cartago,3,6F1D77,FFFFFF +R05,200,San José - Alajuela,3,F5A623,000000 diff --git a/lab-04-data-pipeline/data/sample_gtfs/stop_times.txt b/lab-04-data-pipeline/data/sample_gtfs/stop_times.txt new file mode 100644 index 0000000..f2ca5f2 --- /dev/null +++ b/lab-04-data-pipeline/data/sample_gtfs/stop_times.txt @@ -0,0 +1,125 @@ +trip_id,stop_id,stop_sequence,arrival_time,departure_time +T01_AM1,S001,1,06:00:00,06:00:00 +T01_AM1,S004,2,06:05:00,06:05:00 +T01_AM1,S006,3,06:15:00,06:15:00 +T01_AM1,S007,4,06:20:00,06:20:00 +T01_AM1,S009,5,06:35:00,06:35:00 +T01_AM1,S011,6,06:50:00,06:50:00 +T01_AM2,S011,1,07:10:00,07:10:00 +T01_AM2,S009,2,07:25:00,07:25:00 +T01_AM2,S007,3,07:40:00,07:40:00 +T01_AM2,S006,4,07:45:00,07:45:00 +T01_AM2,S004,5,07:55:00,07:55:00 +T01_AM2,S001,6,08:00:00,08:00:00 +T01_MD1,S001,1,12:00:00,12:00:00 +T01_MD1,S004,2,12:05:00,12:05:00 +T01_MD1,S006,3,12:15:00,12:15:00 +T01_MD1,S007,4,12:20:00,12:20:00 +T01_MD1,S009,5,12:35:00,12:35:00 +T01_MD1,S011,6,12:50:00,12:50:00 +T01_MD2,S011,1,13:10:00,13:10:00 +T01_MD2,S009,2,13:25:00,13:25:00 +T01_MD2,S007,3,13:40:00,13:40:00 +T01_MD2,S006,4,13:45:00,13:45:00 +T01_MD2,S004,5,13:55:00,13:55:00 +T01_MD2,S001,6,14:00:00,14:00:00 +T01_PM1,S001,1,17:00:00,17:00:00 +T01_PM1,S004,2,17:08:00,17:08:00 +T01_PM1,S006,3,17:22:00,17:22:00 +T01_PM1,S007,4,17:28:00,17:28:00 +T01_PM1,S009,5,17:48:00,17:48:00 +T01_PM1,S011,6,18:05:00,18:05:00 +T01_WE1,S001,1,09:00:00,09:00:00 +T01_WE1,S004,2,09:05:00,09:05:00 +T01_WE1,S006,3,09:15:00,09:15:00 +T01_WE1,S007,4,09:20:00,09:20:00 +T01_WE1,S009,5,09:32:00,09:32:00 +T01_WE1,S011,6,09:45:00,09:45:00 +T01_WE2,S011,1,10:05:00,10:05:00 +T01_WE2,S009,2,10:18:00,10:18:00 +T01_WE2,S007,3,10:30:00,10:30:00 +T01_WE2,S006,4,10:35:00,10:35:00 +T01_WE2,S004,5,10:45:00,10:45:00 +T01_WE2,S001,6,10:50:00,10:50:00 +T02_AM1,S001,1,06:15:00,06:15:00 +T02_AM1,S002,2,06:20:00,06:20:00 +T02_AM1,S023,3,06:28:00,06:28:00 +T02_AM1,S013,4,06:42:00,06:42:00 +T02_AM1,S014,5,06:55:00,06:55:00 +T02_AM2,S014,1,07:15:00,07:15:00 +T02_AM2,S013,2,07:28:00,07:28:00 +T02_AM2,S023,3,07:42:00,07:42:00 +T02_AM2,S002,4,07:50:00,07:50:00 +T02_AM2,S001,5,07:55:00,07:55:00 +T02_MD1,S001,1,12:15:00,12:15:00 +T02_MD1,S002,2,12:20:00,12:20:00 +T02_MD1,S023,3,12:28:00,12:28:00 +T02_MD1,S013,4,12:42:00,12:42:00 +T02_MD1,S014,5,12:55:00,12:55:00 +T02_PM1,S001,1,17:15:00,17:15:00 +T02_PM1,S002,2,17:22:00,17:22:00 +T02_PM1,S023,3,17:32:00,17:32:00 +T02_PM1,S013,4,17:50:00,17:50:00 +T02_PM1,S014,5,18:05:00,18:05:00 +T02_WE1,S001,1,09:15:00,09:15:00 +T02_WE1,S002,2,09:20:00,09:20:00 +T02_WE1,S023,3,09:28:00,09:28:00 +T02_WE1,S013,4,09:40:00,09:40:00 +T02_WE1,S014,5,09:52:00,09:52:00 +T03_AM1,S001,1,06:30:00,06:30:00 +T03_AM1,S003,2,06:35:00,06:35:00 +T03_AM1,S014,3,06:55:00,06:55:00 +T03_AM1,S016,4,07:05:00,07:05:00 +T03_AM1,S017,5,07:20:00,07:20:00 +T03_AM2,S017,1,07:40:00,07:40:00 +T03_AM2,S016,2,07:55:00,07:55:00 +T03_AM2,S014,3,08:05:00,08:05:00 +T03_AM2,S003,4,08:25:00,08:25:00 +T03_AM2,S001,5,08:30:00,08:30:00 +T03_MD1,S001,1,12:30:00,12:30:00 +T03_MD1,S003,2,12:35:00,12:35:00 +T03_MD1,S014,3,12:55:00,12:55:00 +T03_MD1,S016,4,13:05:00,13:05:00 +T03_MD1,S017,5,13:20:00,13:20:00 +T03_PM1,S001,1,17:30:00,17:30:00 +T03_PM1,S003,2,17:37:00,17:37:00 +T03_PM1,S014,3,18:00:00,18:00:00 +T03_PM1,S016,4,18:12:00,18:12:00 +T03_PM1,S017,5,18:28:00,18:28:00 +T03_WE1,S001,1,09:30:00,09:30:00 +T03_WE1,S003,2,09:35:00,09:35:00 +T03_WE1,S014,3,09:52:00,09:52:00 +T03_WE1,S016,4,10:02:00,10:02:00 +T03_WE1,S017,5,10:15:00,10:15:00 +T04_AM1,S001,1,05:30:00,05:30:00 +T04_AM1,S018,2,05:58:00,05:58:00 +T04_AM1,S019,3,06:15:00,06:15:00 +T04_AM1,S020,4,06:30:00,06:30:00 +T04_AM2,S020,1,06:45:00,06:45:00 +T04_AM2,S019,2,07:00:00,07:00:00 +T04_AM2,S018,3,07:18:00,07:18:00 +T04_AM2,S001,5,07:48:00,07:48:00 +T04_MD1,S001,1,12:00:00,12:00:00 +T04_MD1,S018,2,12:28:00,12:28:00 +T04_MD1,S019,3,12:45:00,12:45:00 +T04_MD1,S020,4,13:00:00,13:00:00 +T04_PM1,S001,1,17:00:00,17:00:00 +T04_PM1,S018,2,17:35:00,17:35:00 +T04_PM1,S019,3,17:55:00,17:55:00 +T04_PM1,S020,4,18:15:00,18:15:00 +T05_AM1,S001,1,05:45:00,05:45:00 +T05_AM1,S022,2,05:52:00,05:52:00 +T05_AM1,S012,3,06:00:00,06:00:00 +T05_AM1,S021,4,06:30:00,06:30:00 +T05_AM2,S021,1,07:00:00,07:00:00 +T05_AM2,S012,2,07:30:00,07:30:00 +T05_AM2,S022,3,07:38:00,07:38:00 +T05_AM2,S001,4,07:45:00,07:45:00 +T05_MD1,S001,1,12:00:00,12:00:00 +T05_MD1,S022,2,12:07:00,12:07:00 +T05_MD1,S012,3,12:15:00,12:15:00 +T05_MD1,S021,4,12:45:00,12:45:00 +T05_PM1,S001,1,17:00:00,17:00:00 +T05_PM1,S022,2,17:10:00,17:10:00 +T05_PM1,S012,3,17:20:00,17:20:00 +T05_PM1,S021,4,17:55:00,17:55:00 diff --git a/lab-04-data-pipeline/data/sample_gtfs/stops.txt b/lab-04-data-pipeline/data/sample_gtfs/stops.txt new file mode 100644 index 0000000..f7551ce --- /dev/null +++ b/lab-04-data-pipeline/data/sample_gtfs/stops.txt @@ -0,0 +1,24 @@ +stop_id,stop_name,stop_lat,stop_lon,zone_id,wheelchair_boarding +S001,Terminal 7-10 (San José),9.93370,-84.08000,Z1,1 +S002,Mercado Central,9.93070,-84.07950,Z1,1 +S003,Plaza de la Cultura,9.93180,-84.07800,Z1,1 +S004,Hospital San Juan de Dios,9.93120,-84.08550,Z1,1 +S005,Sabana Norte,9.94190,-84.10190,Z1,0 +S006,Parque La Sabana,9.93870,-84.10880,Z1,1 +S007,Estadio Nacional,9.93550,-84.11220,Z2,0 +S008,Pavas Centro,9.93950,-84.13680,Z2,0 +S009,Escazú Centro,9.91930,-84.13670,Z2,1 +S010,Guachipelín,9.91740,-84.15720,Z2,0 +S011,Santa Ana Centro,9.92960,-84.18350,Z3,0 +S012,La Uruca,9.94860,-84.11050,Z1,0 +S013,UCR San Pedro,9.93800,-84.05030,Z1,1 +S014,Curridabat Centro,9.91340,-84.05020,Z2,0 +S015,Pinares,9.90170,-84.04220,Z2,0 +S016,Desamparados Centro,9.89870,-84.06370,Z2,1 +S017,San Miguel de Desamparados,9.87620,-84.05710,Z3,0 +S018,San Diego de La Unión,9.90030,-83.98570,Z3,0 +S019,Taras,9.87930,-83.94930,Z3,0 +S020,Cartago Centro,9.86410,-83.91960,Z4,1 +S021,Alajuela Centro,10.01620,-84.21430,Z3,1 +S022,La Uruca - Intersección,9.94710,-84.11380,Z1,0 +S023,Barrio México,9.93980,-84.09150,Z1,0 diff --git a/lab-04-data-pipeline/data/sample_gtfs/trips.txt b/lab-04-data-pipeline/data/sample_gtfs/trips.txt new file mode 100644 index 0000000..e99f13e --- /dev/null +++ b/lab-04-data-pipeline/data/sample_gtfs/trips.txt @@ -0,0 +1,26 @@ +trip_id,route_id,service_id,trip_headsign,direction_id +T01_AM1,R01,WEEKDAY,Escazú Centro,0 +T01_AM2,R01,WEEKDAY,San José Terminal 7-10,1 +T01_MD1,R01,WEEKDAY,Escazú Centro,0 +T01_MD2,R01,WEEKDAY,San José Terminal 7-10,1 +T01_PM1,R01,WEEKDAY,Escazú Centro,0 +T01_WE1,R01,WEEKEND,Escazú Centro,0 +T01_WE2,R01,WEEKEND,San José Terminal 7-10,1 +T02_AM1,R02,WEEKDAY,UCR San Pedro,0 +T02_AM2,R02,WEEKDAY,San José Terminal 7-10,1 +T02_MD1,R02,WEEKDAY,UCR San Pedro,0 +T02_PM1,R02,WEEKDAY,UCR San Pedro,0 +T02_WE1,R02,WEEKEND,UCR San Pedro,0 +T03_AM1,R03,WEEKDAY,Desamparados,0 +T03_AM2,R03,WEEKDAY,San José Terminal 7-10,1 +T03_MD1,R03,WEEKDAY,Desamparados,0 +T03_PM1,R03,WEEKDAY,Desamparados,0 +T03_WE1,R03,WEEKEND,Desamparados,0 +T04_AM1,R04,DAILY,Cartago Centro,0 +T04_AM2,R04,DAILY,San José Terminal 7-10,1 +T04_MD1,R04,DAILY,Cartago Centro,0 +T04_PM1,R04,DAILY,Cartago Centro,0 +T05_AM1,R05,DAILY,Alajuela Centro,0 +T05_AM2,R05,DAILY,San José Terminal 7-10,1 +T05_MD1,R05,DAILY,Alajuela Centro,0 +T05_PM1,R05,DAILY,Alajuela Centro,0 diff --git a/lab-04-data-pipeline/docker-compose.yml b/lab-04-data-pipeline/docker-compose.yml new file mode 100644 index 0000000..e61c751 --- /dev/null +++ b/lab-04-data-pipeline/docker-compose.yml @@ -0,0 +1,65 @@ +services: + + db: + image: timescale/timescaledb:latest-pg16 + environment: + POSTGRES_DB: simovi_pipeline + POSTGRES_USER: simovi + POSTGRES_PASSWORD: simovi_pass + ports: + - "5433:5432" + volumes: + - timescale_data:/var/lib/postgresql/data + healthcheck: + test: ["CMD-SHELL", "pg_isready -U simovi -d simovi_pipeline"] + interval: 5s + timeout: 5s + retries: 10 + + prefect-server: + image: prefecthq/prefect:3-latest + command: prefect server start --host 0.0.0.0 + ports: + - "4200:4200" + environment: + PREFECT_SERVER_API_HOST: 0.0.0.0 + PREFECT_UI_API_URL: http://localhost:4200/api + healthcheck: + test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:4200/api/health')"] + interval: 10s + timeout: 5s + retries: 12 + + prefect-worker: + build: . + command: > + sh -c "prefect worker start --pool simovi-pool --type process" + environment: + PREFECT_API_URL: http://prefect-server:4200/api + DATABASE_URL: postgresql://simovi:simovi_pass@db:5432/simovi_pipeline + volumes: + - .:/app + depends_on: + prefect-server: + condition: service_healthy + db: + condition: service_healthy + + app: + build: . + command: sleep infinity + environment: + PREFECT_API_URL: http://prefect-server:4200/api + DATABASE_URL: postgresql://simovi:simovi_pass@db:5432/simovi_pipeline + ports: + - "8888:8888" + volumes: + - .:/app + depends_on: + prefect-server: + condition: service_healthy + db: + condition: service_healthy + +volumes: + timescale_data: diff --git a/lab-04-data-pipeline/flows/analyze.py b/lab-04-data-pipeline/flows/analyze.py new file mode 100644 index 0000000..b1ea9e9 --- /dev/null +++ b/lab-04-data-pipeline/flows/analyze.py @@ -0,0 +1,232 @@ +""" +Flow: analyze +Demuestra las capacidades de series temporales de TimescaleDB. + +Etapas +------ +1. seed_vehicle_events — genera eventos simulados + de vehículos (últimas 2 horas) +2. seed_stop_ridership — genera conteos simulados de pasajeros +3. query_speed_buckets — velocidad promedio por vehículo en ventanas de 5 min +4. query_peak_stops — paradas con más abordajes en la última hora +5. query_fleet_summary — resumen de flota: vehículos activos, velocidad media +""" + +import os +import random +from datetime import datetime, timedelta, timezone + +import polars as pl +import psycopg2 +import psycopg2.extras +from prefect import flow, task, get_run_logger + +VEHICLE_IDS = ["BUS-101", "BUS-102", + "BUS-201", "BUS-301", "BUS-401", "BUS-501"] + +# Coordenadas aproximadas del centro de San José para la simulación +BASE_LAT = 9.9337 +BASE_LON = -84.0800 + + +def get_connection() -> psycopg2.extensions.connection: + return psycopg2.connect(os.environ["DATABASE_URL"]) + + +# ── Seed tasks ────────────────────────────────────────────────────────────── + +@task(name="seed-vehicle-events") +def seed_vehicle_events(hours_back: int = 2, + events_per_vehicle: int = 24) -> int: + """ + Inserta eventos de posición simulados en la hypertable vehicle_events. + Cada vehículo genera `events_per_vehicle` eventos distribuidos en + los últimos `hours_back` horas. + """ + logger = get_run_logger() + now = datetime.now(tz=timezone.utc) + interval = timedelta(hours=hours_back) / events_per_vehicle + + rows = [] + for vid in VEHICLE_IDS: + lat, lon = BASE_LAT, BASE_LON + for i in range(events_per_vehicle): + recorded_at = now - timedelta(hours=hours_back) + interval * i + # Movimiento aleatorio pequeño (≈ 200 m por paso) + lat += random.uniform(-0.001, 0.001) + lon += random.uniform(-0.001, 0.001) + speed = random.uniform(0, 60) + heading = random.uniform(0, 360) + rows.append((recorded_at, vid, lat, lon, speed, heading)) + + sql = """ + INSERT INTO vehicle_events + (recorded_at, vehicle_id, latitude, longitude, speed_kmh, heading) + VALUES %s + """ + with get_connection() as conn: + with conn.cursor() as cur: + psycopg2.extras.execute_values(cur, sql, rows) + conn.commit() + + logger.info(f" vehicle_events: {len(rows)} eventos simulados insertados.") + return len(rows) + + +@task(name="seed-stop-ridership") +def seed_stop_ridership(hours_back: int = 2, + stop_ids: list[str] | None = None) -> int: + """ + Inserta conteos de pasajeros simulados en la hypertable stop_ridership. + Se generan registros cada 15 minutos para cada parada. + """ + logger = get_run_logger() + + # Obtener stop_ids desde la BD si no se pasan explícitamente + if stop_ids is None: + with get_connection() as conn: + with conn.cursor() as cur: + cur.execute("SELECT stop_id FROM stops LIMIT 10") + stop_ids = [r[0] for r in cur.fetchall()] + + now = datetime.now(tz=timezone.utc) + rows = [] + for sid in stop_ids: + ts = now - timedelta(hours=hours_back) + while ts < now: + boardings = random.randint(0, 30) + alightings = random.randint(0, 30) + rows.append((ts, sid, "R01", boardings, alightings)) + ts += timedelta(minutes=15) + + sql = """ + INSERT INTO stop_ridership + (recorded_at, stop_id, route_id, boardings, alightings) + VALUES %s + """ + with get_connection() as conn: + with conn.cursor() as cur: + psycopg2.extras.execute_values(cur, sql, rows) + conn.commit() + + logger.info(" stop_ridership: " + f"{len(rows)} registros simulados insertados.") + return len(rows) + + +# ── Analysis tasks ─────────────────────────────────────────────────────────── + +@task(name="query-speed-buckets") +def query_speed_buckets(window_minutes: int = 5) -> pl.DataFrame: + """ + Velocidad promedio por vehículo en ventanas de `window_minutes` minutos + usando la función time_bucket() de TimescaleDB. + """ + logger = get_run_logger() + sql = f""" + SELECT + time_bucket('{window_minutes} minutes', recorded_at) AS bucket, + vehicle_id, + ROUND(AVG(speed_kmh)::NUMERIC, 1) AS avg_speed_kmh, + COUNT(*) AS samples + FROM vehicle_events + WHERE recorded_at > NOW() - INTERVAL '2 hours' + GROUP BY bucket, vehicle_id + ORDER BY bucket DESC, vehicle_id + LIMIT 30 + """ + with get_connection() as conn: + with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: + cur.execute(sql) + rows = cur.fetchall() + + df = pl.DataFrame([dict(r) for r in rows]) + logger.info(f" speed_buckets: {df.shape[0]} filas") + logger.info(f"\n{df}") + return df + + +@task(name="query-peak-stops") +def query_peak_stops(top_n: int = 5) -> pl.DataFrame: + """ + Paradas con mayor cantidad de abordajes en la última hora. + Combina la hypertable stop_ridership con la tabla de referencia stops. + """ + logger = get_run_logger() + sql = f""" + SELECT + sr.stop_id, + s.stop_name, + SUM(sr.boardings) AS total_boardings, + SUM(sr.alightings) AS total_alightings + FROM stop_ridership sr + JOIN stops s USING (stop_id) + WHERE sr.recorded_at > NOW() - INTERVAL '1 hour' + GROUP BY sr.stop_id, s.stop_name + ORDER BY total_boardings DESC + LIMIT {top_n} + """ + with get_connection() as conn: + with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: + cur.execute(sql) + rows = cur.fetchall() + + df = pl.DataFrame([dict(r) for r in rows]) + logger.info(f" peak_stops (top {top_n}):\n{df}") + return df + + +@task(name="query-fleet-summary") +def query_fleet_summary() -> dict: + """ + Resumen de flota en los últimos 30 minutos: + vehículos activos, velocidad media y máxima. + """ + logger = get_run_logger() + sql = """ + SELECT + COUNT(DISTINCT vehicle_id) AS active_vehicles, + ROUND(AVG(speed_kmh)::NUMERIC, 1) AS avg_speed_kmh, + ROUND(MAX(speed_kmh)::NUMERIC, 1) AS max_speed_kmh + FROM vehicle_events + WHERE recorded_at > NOW() - INTERVAL '30 minutes' + """ + with get_connection() as conn: + with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: + cur.execute(sql) + row = cur.fetchone() + + summary = dict(row) + logger.info(f" fleet_summary: {summary}") + return summary + + +# ── Flow ──────────────────────────────────────────────────────────────────── + +@flow(name="analyze-timescale", log_prints=True) +def analyze_timescale() -> dict: + """ + Popula las hypertables con datos simulados y ejecuta queries de análisis + que demuestran las capacidades de series temporales de TimescaleDB. + """ + logger = get_run_logger() + logger.info("=== Iniciando análisis TimescaleDB ===") + + seed_vehicle_events() + seed_stop_ridership() + + speed_df = query_speed_buckets(window_minutes=5) + peak_df = query_peak_stops(top_n=5) + summary = query_fleet_summary() + + results = { + "speed_buckets_rows": speed_df.shape[0], + "peak_stops_rows": peak_df.shape[0], + "fleet_summary": summary, + } + logger.info(f"=== Análisis completado: {results} ===") + return results + + +if __name__ == "__main__": + analyze_timescale() diff --git a/lab-04-data-pipeline/flows/ingest_gtfs.py b/lab-04-data-pipeline/flows/ingest_gtfs.py new file mode 100644 index 0000000..181eb7a --- /dev/null +++ b/lab-04-data-pipeline/flows/ingest_gtfs.py @@ -0,0 +1,279 @@ +""" +Flow: ingest_gtfs +Lectura de archivos GTFS → validación de schema → carga a TimescaleDB. + +Etapas +------ +1. apply_schema — crea tablas e hypertables si no existen +2. read_gtfs_file — lee cada .txt con Polars +3. validate_file — comprueba columnas obligatorias +4. load_* — inserta en la tabla correspondiente con COPY (bulk insert) +""" + +import os +from pathlib import Path + +import polars as pl +import psycopg2 +import psycopg2.extras +from prefect import flow, task, get_run_logger + +GTFS_DIR = Path(__file__).parent.parent / "data" / "sample_gtfs" +SCHEMA_FILE = Path(__file__).parent.parent / "models" / "timescale_schema.sql" + +REQUIRED_COLUMNS = { + "stops": {"stop_id", "stop_name", "stop_lat", "stop_lon"}, + "routes": {"route_id", "route_short_name", "route_long_name", "route_type"}, + "trips": {"trip_id", "route_id", "service_id"}, + "stop_times": {"trip_id", "stop_id", "stop_sequence", "arrival_time", "departure_time"}, + "calendar": { + "service_id", "monday", "tuesday", "wednesday", + "thursday", "friday", "saturday", "sunday", + "start_date", "end_date", + }, +} + + +def get_connection() -> psycopg2.extensions.connection: + return psycopg2.connect(os.environ["DATABASE_URL"]) + + +# ── Tasks ──────────────────────────────────────────────────────────────────── + +@task(name="apply-schema", retries=2, retry_delay_seconds=5) +def apply_schema() -> None: + logger = get_run_logger() + sql = SCHEMA_FILE.read_text() + with get_connection() as conn: + conn.autocommit = True + with conn.cursor() as cur: + cur.execute(sql) + logger.info("Schema aplicado correctamente.") + + +@task(name="read-gtfs-file") +def read_gtfs_file(name: str) -> pl.DataFrame: + """Lee un archivo GTFS (.txt) con Polars y retorna un DataFrame.""" + logger = get_run_logger() + path = GTFS_DIR / f"{name}.txt" + # todo como texto inicialmente + df = pl.read_csv(path, infer_schema_length=0) + logger.info(f" {name}.txt → {df.shape[0]} filas, {df.shape[1]} columnas") + return df + + +@task(name="validate-gtfs-file") +def validate_file(df: pl.DataFrame, name: str) -> pl.DataFrame: + """Verifica que el DataFrame contenga + las columnas obligatorias del spec GTFS.""" + logger = get_run_logger() + required = REQUIRED_COLUMNS[name] + actual = set(df.columns) + missing = required - actual + if missing: + raise ValueError(f"{name}.txt faltan columnas: {missing}") + logger.info(f" {name}.txt validado — columnas OK.") + return df + + +@task(name="load-stops", retries=1) +def load_stops(df: pl.DataFrame) -> int: + logger = get_run_logger() + rows = [ + ( + row["stop_id"], + row["stop_name"], + float(row["stop_lat"]), + float(row["stop_lon"]), + row.get("zone_id"), + int(row["wheelchair_boarding"]) if row.get("wheelchair_boarding") else 0, + ) + for row in df.to_dicts() + ] + sql = """ + INSERT INTO stops (stop_id, stop_name, stop_lat, stop_lon, zone_id, wheelchair) + VALUES %s + ON CONFLICT (stop_id) DO UPDATE SET + stop_name = EXCLUDED.stop_name, + stop_lat = EXCLUDED.stop_lat, + stop_lon = EXCLUDED.stop_lon, + zone_id = EXCLUDED.zone_id, + wheelchair = EXCLUDED.wheelchair + """ + with get_connection() as conn: + with conn.cursor() as cur: + psycopg2.extras.execute_values(cur, sql, rows) + conn.commit() + logger.info(f" stops: {len(rows)} filas insertadas / actualizadas.") + return len(rows) + + +@task(name="load-routes", retries=1) +def load_routes(df: pl.DataFrame) -> int: + logger = get_run_logger() + rows = [ + ( + row["route_id"], + row["route_short_name"], + row["route_long_name"], + int(row["route_type"]), + row.get("route_color"), + row.get("route_text_color"), + ) + for row in df.to_dicts() + ] + sql = """ + INSERT INTO routes + (route_id, route_short_name, route_long_name, route_type, + route_color, route_text_color) + VALUES %s + ON CONFLICT (route_id) DO UPDATE SET + route_short_name = EXCLUDED.route_short_name, + route_long_name = EXCLUDED.route_long_name, + route_type = EXCLUDED.route_type, + route_color = EXCLUDED.route_color, + route_text_color = EXCLUDED.route_text_color + """ + with get_connection() as conn: + with conn.cursor() as cur: + psycopg2.extras.execute_values(cur, sql, rows) + conn.commit() + logger.info(f" routes: {len(rows)} filas insertadas / actualizadas.") + return len(rows) + + +@task(name="load-calendar", retries=1) +def load_calendar(df: pl.DataFrame) -> int: + logger = get_run_logger() + bool_cols = ["monday", "tuesday", "wednesday", "thursday", + "friday", "saturday", "sunday"] + rows = [ + ( + row["service_id"], + *[row[c] == "1" for c in bool_cols], + row["start_date"], + row["end_date"], + ) + for row in df.to_dicts() + ] + sql = """ + INSERT INTO calendar + (service_id, monday, tuesday, wednesday, thursday, + friday, saturday, sunday, start_date, end_date) + VALUES %s + ON CONFLICT (service_id) DO NOTHING + """ + with get_connection() as conn: + with conn.cursor() as cur: + psycopg2.extras.execute_values(cur, sql, rows) + conn.commit() + logger.info(f" calendar: {len(rows)} filas insertadas.") + return len(rows) + + +@task(name="load-trips", retries=1) +def load_trips(df: pl.DataFrame) -> int: + logger = get_run_logger() + rows = [ + ( + row["trip_id"], + row["route_id"], + row["service_id"], + row.get("trip_headsign"), + int(row["direction_id"]) if row.get("direction_id") else None, + ) + for row in df.to_dicts() + ] + sql = """ + INSERT INTO trips (trip_id, route_id, service_id, trip_headsign, direction_id) + VALUES %s + ON CONFLICT (trip_id) DO UPDATE SET + trip_headsign = EXCLUDED.trip_headsign, + direction_id = EXCLUDED.direction_id + """ + with get_connection() as conn: + with conn.cursor() as cur: + psycopg2.extras.execute_values(cur, sql, rows) + conn.commit() + logger.info(f" trips: {len(rows)} filas insertadas / actualizadas.") + return len(rows) + + +@task(name="load-stop-times", retries=1) +def load_stop_times(df: pl.DataFrame) -> int: + logger = get_run_logger() + rows = [ + ( + row["trip_id"], + row["stop_id"], + int(row["stop_sequence"]), + row["arrival_time"], + row["departure_time"], + ) + for row in df.to_dicts() + ] + sql = """ + INSERT INTO stop_times + (trip_id, stop_id, stop_sequence, arrival_time, departure_time) + VALUES %s + ON CONFLICT (trip_id, stop_sequence) DO UPDATE SET + stop_id = EXCLUDED.stop_id, + arrival_time = EXCLUDED.arrival_time, + departure_time = EXCLUDED.departure_time + """ + with get_connection() as conn: + with conn.cursor() as cur: + psycopg2.extras.execute_values(cur, sql, rows) + conn.commit() + logger.info(f" stop_times: {len(rows)} filas insertadas / actualizadas.") + return len(rows) + + +# ── Flow ───────────────────────────────────────────────────────────────────── + +@flow(name="ingest-gtfs", log_prints=True) +def ingest_gtfs(gtfs_dir: str | None = None) -> dict[str, int]: + """ + Pipeline principal de ingesta GTFS. + + Orden de carga respeta las foreign keys: + stops → routes → calendar → trips → stop_times + """ + logger = get_run_logger() + logger.info("=== Iniciando ingesta GTFS ===") + + apply_schema() + + # Lectura y validación en paralelo (sin dependencias entre sí) + stops_df = read_gtfs_file("stops") + routes_df = read_gtfs_file("routes") + calendar_df = read_gtfs_file("calendar") + trips_df = read_gtfs_file("trips") + stop_times_df = read_gtfs_file("stop_times") + + stops_df = validate_file(stops_df, "stops") + routes_df = validate_file(routes_df, "routes") + calendar_df = validate_file(calendar_df, "calendar") + trips_df = validate_file(trips_df, "trips") + stop_times_df = validate_file(stop_times_df, "stop_times") + + # Carga en orden (respeta FK) + n_stops = load_stops(stops_df) + n_routes = load_routes(routes_df) + n_calendar = load_calendar(calendar_df) + n_trips = load_trips(trips_df) + n_stop_times = load_stop_times(stop_times_df) + + summary = { + "stops": n_stops, + "routes": n_routes, + "calendar": n_calendar, + "trips": n_trips, + "stop_times": n_stop_times, + } + logger.info(f"=== Ingesta completada: {summary} ===") + return summary + + +if __name__ == "__main__": + ingest_gtfs() diff --git a/lab-04-data-pipeline/flows/transform.py b/lab-04-data-pipeline/flows/transform.py new file mode 100644 index 0000000..17bb114 --- /dev/null +++ b/lab-04-data-pipeline/flows/transform.py @@ -0,0 +1,207 @@ +""" +Flow: transform +Transformaciones con Polars sobre los datos GTFS ya ingestados. + +Produce +------- +- Estadísticas por ruta: cantidad de paradas, viajes y frecuencia de servicio +- Paradas más concurridas (cantidad de viajes que las sirven) +- Exportación a Parquet en data/processed/ +""" + +import os +from pathlib import Path + +import polars as pl +import psycopg2 +import psycopg2.extras +from prefect import flow, task, get_run_logger + +OUTPUT_DIR = Path(__file__).parent.parent / "data" / "processed" + + +def get_connection() -> psycopg2.extensions.connection: + return psycopg2.connect(os.environ["DATABASE_URL"]) + + +# ── Tasks ──────────────────────────────────────────────────────────────────── + +@task(name="fetch-table") +def fetch_table(table: str) -> pl.DataFrame: + """Lee una tabla completa de TimescaleDB + y la retorna como DataFrame de Polars.""" + logger = get_run_logger() + with get_connection() as conn: + with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: + cur.execute(f"SELECT * FROM {table}") # noqa: S608 + rows = cur.fetchall() + df = pl.DataFrame([dict(r) for r in rows]) + logger.info(f" {table}: {df.shape[0]} filas leídas.") + return df + + +@task(name="compute-route-stats") +def compute_route_stats( + routes: pl.DataFrame, + trips: pl.DataFrame, + stop_times: pl.DataFrame, +) -> pl.DataFrame: + """ + Por cada ruta calcula: + - total_trips : cantidad de viajes programados + - total_stops : paradas únicas que sirve la ruta + - avg_stops_trip : promedio de paradas por viaje + """ + logger = get_run_logger() + + # Paradas por viaje + stops_per_trip = ( + stop_times + .group_by("trip_id") + .agg(pl.len().alias("stop_count")) + ) + + # Unir trips → stops_per_trip → routes + trips_with_stops = trips.join(stops_per_trip, on="trip_id", how="left") + + route_stats = ( + trips_with_stops + .group_by("route_id") + .agg( + pl.len().alias("total_trips"), + pl.col("stop_count").sum().alias("total_stop_visits"), + pl.col("stop_count").mean().round(1).alias("avg_stops_trip"), + ) + .join( + routes.select(["route_id", "route_short_name", "route_long_name"]), + on="route_id", + ) + .sort("total_trips", descending=True) + ) + + logger.info(f" route_stats: {route_stats.shape[0]} rutas procesadas.") + return route_stats + + +@task(name="compute-busiest-stops") +def compute_busiest_stops( + stops: pl.DataFrame, + stop_times: pl.DataFrame, +) -> pl.DataFrame: + """ + Paradas ordenadas por cantidad de visitas de viajes (trips que las sirven). + Útil para identificar nodos de alta demanda en la red. + """ + logger = get_run_logger() + + visit_counts = ( + stop_times + .group_by("stop_id") + .agg(pl.len().alias("trip_visits")) + ) + + busiest = ( + stops + .join(visit_counts, on="stop_id", how="left") + .with_columns(pl.col("trip_visits").fill_null(0)) + .sort("trip_visits", descending=True) + .select(["stop_id", "stop_name", "stop_lat", "stop_lon", + "zone_id", "trip_visits"]) + ) + + logger.info( + f" busiest_stops: top parada = " + f"{busiest[0, 'stop_name']} ({busiest[0, 'trip_visits']} visitas)" + ) + return busiest + + +@task(name="compute-service-frequency") +def compute_service_frequency( + trips: pl.DataFrame, + stop_times: pl.DataFrame, +) -> pl.DataFrame: + """ + Frecuencia de servicio por ruta y dirección: + cantidad de viajes por franja horaria (mañana / mediodía / tarde). + """ + logger = get_run_logger() + + # Hora de salida = hora del primer stop_sequence del viaje + first_departure = ( + stop_times + .sort("stop_sequence") + .group_by("trip_id") + .agg(pl.col("departure_time").first().alias("departure_time")) + ) + + # Extraer hora (HH de "HH:MM:SS") + first_departure = first_departure.with_columns( + pl.col("departure_time").str.slice(0, 2).cast(pl.Int32).alias("hour") + ) + + trips_with_hour = trips.join(first_departure, on="trip_id", how="left") + + frequency = ( + trips_with_hour + .with_columns( + pl.when(pl.col("hour") < 12) + .then(pl.lit("mañana")) + .when(pl.col("hour") < 17) + .then(pl.lit("mediodía")) + .otherwise(pl.lit("tarde")) + .alias("period") + ) + .group_by(["route_id", "direction_id", "period"]) + .agg(pl.len().alias("trip_count")) + .sort(["route_id", "direction_id", "period"]) + ) + + logger.info(f" service_frequency: {frequency.shape[0]} filas.") + return frequency + + +@task(name="export-parquet") +def export_parquet(df: pl.DataFrame, name: str) -> Path: + """Exporta un DataFrame a Parquet en data/processed/.""" + logger = get_run_logger() + OUTPUT_DIR.mkdir(parents=True, exist_ok=True) + path = OUTPUT_DIR / f"{name}.parquet" + df.write_parquet(path) + logger.info(f" Exportado: {path} ({path.stat().st_size // 1024} KB)") + return path + + +# ── Flow ───────────────────────────────────────────────────────────────────── + +@flow(name="transform-gtfs", log_prints=True) +def transform_gtfs() -> dict[str, str]: + """ + Transforma los datos GTFS ya cargados en TimescaleDB y exporta + estadísticas en formato Parquet para análisis posterior. + """ + logger = get_run_logger() + logger.info("=== Iniciando transformación GTFS ===") + + routes = fetch_table("routes") + trips = fetch_table("trips") + stops = fetch_table("stops") + stop_times = fetch_table("stop_times") + + route_stats = compute_route_stats(routes, trips, stop_times) + busiest_stops = compute_busiest_stops(stops, stop_times) + service_freq = compute_service_frequency(trips, stop_times) + + paths = { + "route_stats": str(export_parquet(route_stats, "route_stats")), + "busiest_stops": str(export_parquet(busiest_stops, "busiest_stops")), + "service_frequency": str(export_parquet(service_freq, + "service_frequency")), + } + + logger.info(f"=== Transformación completada. Archivos: {paths} ===") + return paths + + +if __name__ == "__main__": + transform_gtfs() diff --git a/lab-04-data-pipeline/models/timescale_schema.sql b/lab-04-data-pipeline/models/timescale_schema.sql new file mode 100644 index 0000000..d2bb25e --- /dev/null +++ b/lab-04-data-pipeline/models/timescale_schema.sql @@ -0,0 +1,122 @@ +-- ============================================================ +-- Lab 04 — TimescaleDB Schema +-- Dominio: SIMOVI — Sistema de Monitoreo de Vehículos +-- ============================================================ + +-- Extensión requerida por TimescaleDB +CREATE EXTENSION IF NOT EXISTS timescaledb; + +-- ============================================================ +-- Tablas normales (datos de referencia GTFS) +-- ============================================================ + +CREATE TABLE IF NOT EXISTS stops ( + stop_id TEXT PRIMARY KEY, + stop_name TEXT NOT NULL, + stop_lat DOUBLE PRECISION NOT NULL, + stop_lon DOUBLE PRECISION NOT NULL, + zone_id TEXT, + wheelchair SMALLINT DEFAULT 0 -- 0=sin info, 1=accesible, 2=no accesible +); + +CREATE TABLE IF NOT EXISTS routes ( + route_id TEXT PRIMARY KEY, + route_short_name TEXT NOT NULL, + route_long_name TEXT NOT NULL, + route_type SMALLINT NOT NULL, -- 3 = bus + route_color TEXT, + route_text_color TEXT +); + +CREATE TABLE IF NOT EXISTS trips ( + trip_id TEXT PRIMARY KEY, + route_id TEXT NOT NULL REFERENCES routes(route_id), + service_id TEXT NOT NULL, + trip_headsign TEXT, + direction_id SMALLINT -- 0 = ida, 1 = vuelta +); + +CREATE TABLE IF NOT EXISTS stop_times ( + trip_id TEXT NOT NULL REFERENCES trips(trip_id), + stop_id TEXT NOT NULL REFERENCES stops(stop_id), + stop_sequence INTEGER NOT NULL, + arrival_time TEXT NOT NULL, -- HH:MM:SS (puede superar 24h en GTFS) + departure_time TEXT NOT NULL, + PRIMARY KEY (trip_id, stop_sequence) +); + +CREATE TABLE IF NOT EXISTS calendar ( + service_id TEXT PRIMARY KEY, + monday BOOLEAN NOT NULL, + tuesday BOOLEAN NOT NULL, + wednesday BOOLEAN NOT NULL, + thursday BOOLEAN NOT NULL, + friday BOOLEAN NOT NULL, + saturday BOOLEAN NOT NULL, + sunday BOOLEAN NOT NULL, + start_date DATE NOT NULL, + end_date DATE NOT NULL +); + +-- ============================================================ +-- Hypertables (series temporales) +-- ============================================================ + +-- Eventos de posición de vehículos en tiempo real. +-- Equivalente a lo que ingería el MQTT consumer en Lab 03, +-- pero persistido en TimescaleDB para análisis histórico. +CREATE TABLE IF NOT EXISTS vehicle_events ( + recorded_at TIMESTAMPTZ NOT NULL, + vehicle_id TEXT NOT NULL, + trip_id TEXT, + stop_id TEXT, + latitude DOUBLE PRECISION NOT NULL, + longitude DOUBLE PRECISION NOT NULL, + speed_kmh REAL, + heading REAL, + occupancy SMALLINT -- 0=vacío … 5=lleno +); + +SELECT create_hypertable( + 'vehicle_events', + 'recorded_at', + if_not_exists => TRUE, + chunk_time_interval => INTERVAL '1 day' +); + +CREATE INDEX IF NOT EXISTS idx_vehicle_events_vehicle_id + ON vehicle_events (vehicle_id, recorded_at DESC); + +-- Conteo de pasajeros por parada (integración con torniquetes / datos simulados). +CREATE TABLE IF NOT EXISTS stop_ridership ( + recorded_at TIMESTAMPTZ NOT NULL, + stop_id TEXT NOT NULL, + route_id TEXT NOT NULL, + boardings INTEGER NOT NULL DEFAULT 0, + alightings INTEGER NOT NULL DEFAULT 0 +); + +SELECT create_hypertable( + 'stop_ridership', + 'recorded_at', + if_not_exists => TRUE, + chunk_time_interval => INTERVAL '1 day' +); + +CREATE INDEX IF NOT EXISTS idx_stop_ridership_stop_id + ON stop_ridership (stop_id, recorded_at DESC); + +-- ============================================================ +-- Vista: velocidad promedio por vehículo en ventanas de 5 min +-- Ejemplo de uso de time_bucket() +-- ============================================================ + +CREATE OR REPLACE VIEW vehicle_speed_buckets AS +SELECT + time_bucket('5 minutes', recorded_at) AS bucket, + vehicle_id, + ROUND(AVG(speed_kmh)::NUMERIC, 1) AS avg_speed_kmh, + COUNT(*) AS sample_count +FROM vehicle_events +GROUP BY bucket, vehicle_id +ORDER BY bucket DESC, vehicle_id; diff --git a/lab-04-data-pipeline/notebooks/explore_data.ipynb b/lab-04-data-pipeline/notebooks/explore_data.ipynb new file mode 100644 index 0000000..2127215 --- /dev/null +++ b/lab-04-data-pipeline/notebooks/explore_data.ipynb @@ -0,0 +1,400 @@ +{ + "cells": [ + { + "cell_type": "markdown", + "id": "a1b2c3d4", + "metadata": {}, + "source": [ + "# Lab 04 - Exploración de datos GTFS\n", + "\n", + "Análisis descriptivo del feed GTFS cargado en TimescaleDB.\n", + "Consume los Parquet de flows/transform.py y queries directas a TimescaleDB.\n", + "\n", + "**Pre-requisitos**\n", + "docker compose exec app python flows/ingest_gtfs.py\n", + "docker compose exec app python flows/transform.py\n", + "docker compose exec app python flows/analyze.py" + ] + }, + { + "cell_type": "code", + "execution_count": 9, + "id": "b2c3d4e5", + "metadata": {}, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "Conexion OK - postgresql://simovi:simovi_pass@db:5432/simovi_pipeline\n" + ] + } + ], + "source": [ + "import os\n", + "from pathlib import Path\n", + "import polars as pl\n", + "import psycopg2\n", + "import psycopg2.extras\n", + "\n", + "DB_URL = os.environ[\"DATABASE_URL\"]\n", + "PROCESSED = Path(\"..\") / \"data\" / \"processed\"\n", + "\n", + "def query(sql):\n", + " with psycopg2.connect(DB_URL) as conn:\n", + " with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:\n", + " cur.execute(sql)\n", + " rows = cur.fetchall()\n", + " if not rows:\n", + " cols = [d[0] for d in cur.description] if cur.description else []\n", + " return pl.DataFrame({c: [] for c in cols})\n", + " return pl.DataFrame([dict(r) for r in rows])\n", + "\n", + "print(\"Conexion OK -\", DB_URL)" + ] + }, + { + "cell_type": "markdown", + "id": "c3d4e5f6", + "metadata": {}, + "source": [ + "## 1. Resumen del feed GTFS" + ] + }, + { + "cell_type": "code", + "execution_count": 10, + "id": "d4e5f6a7", + "metadata": {}, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "shape: (6, 2)\n", + "┌────────────────┬───────┐\n", + "│ entity ┆ total │\n", + "│ --- ┆ --- │\n", + "│ str ┆ i64 │\n", + "╞════════════════╪═══════╡\n", + "│ stops ┆ 0 │\n", + "│ routes ┆ 0 │\n", + "│ trips ┆ 0 │\n", + "│ stop_times ┆ 0 │\n", + "│ vehicle_events ┆ 0 │\n", + "│ stop_ridership ┆ 0 │\n", + "└────────────────┴───────┘\n" + ] + } + ], + "source": [ + "counts = query(\n", + " \"SELECT 'stops' AS entity, COUNT(*) AS total FROM stops\"\n", + " \" UNION ALL SELECT 'routes', COUNT(*) FROM routes\"\n", + " \" UNION ALL SELECT 'trips', COUNT(*) FROM trips\"\n", + " \" UNION ALL SELECT 'stop_times', COUNT(*) FROM stop_times\"\n", + " \" UNION ALL SELECT 'vehicle_events', COUNT(*) FROM vehicle_events\"\n", + " \" UNION ALL SELECT 'stop_ridership', COUNT(*) FROM stop_ridership\"\n", + ")\n", + "print(counts)" + ] + }, + { + "cell_type": "markdown", + "id": "e5f6a7b8", + "metadata": {}, + "source": [ + "## 2. Estadisticas por ruta (desde Parquet)" + ] + }, + { + "cell_type": "code", + "execution_count": 11, + "id": "f6a7b8c9", + "metadata": {}, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "shape: (5, 4)\n", + "┌──────────────────┬─────────────────────────────────┬─────────────┬────────────────┐\n", + "│ route_short_name ┆ route_long_name ┆ total_trips ┆ avg_stops_trip │\n", + "│ --- ┆ --- ┆ --- ┆ --- │\n", + "│ str ┆ str ┆ u32 ┆ f64 │\n", + "╞══════════════════╪═════════════════════════════════╪═════════════╪════════════════╡\n", + "│ 102 ┆ San José - Escazú vía La Saban… ┆ 7 ┆ 6.0 │\n", + "│ 328 ┆ San José - San Pedro vía UCR ┆ 5 ┆ 5.0 │\n", + "│ 401 ┆ San José - Desamparados ┆ 5 ┆ 5.0 │\n", + "│ 310 ┆ San José - Cartago ┆ 4 ┆ 4.0 │\n", + "│ 200 ┆ San José - Alajuela ┆ 4 ┆ 4.0 │\n", + "└──────────────────┴─────────────────────────────────┴─────────────┴────────────────┘\n" + ] + } + ], + "source": [ + "route_stats = pl.read_parquet(PROCESSED / \"route_stats.parquet\")\n", + "print(route_stats.select([\"route_short_name\",\"route_long_name\",\"total_trips\",\"avg_stops_trip\"]))" + ] + }, + { + "cell_type": "markdown", + "id": "a7b8c9d0", + "metadata": {}, + "source": [ + "## 3. Paradas mas concurridas (desde Parquet)" + ] + }, + { + "cell_type": "code", + "execution_count": 12, + "id": "b8c9d0e1", + "metadata": {}, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "shape: (10, 3)\n", + "┌───────────────────────────┬─────────┬─────────────┐\n", + "│ stop_name ┆ zone_id ┆ trip_visits │\n", + "│ --- ┆ --- ┆ --- │\n", + "│ str ┆ str ┆ u32 │\n", + "╞═══════════════════════════╪═════════╪═════════════╡\n", + "│ Terminal 7-10 (San José) ┆ Z1 ┆ 25 │\n", + "│ Curridabat Centro ┆ Z2 ┆ 10 │\n", + "│ Hospital San Juan de Dios ┆ Z1 ┆ 7 │\n", + "│ Parque La Sabana ┆ Z1 ┆ 7 │\n", + "│ Estadio Nacional ┆ Z2 ┆ 7 │\n", + "│ Escazú Centro ┆ Z2 ┆ 7 │\n", + "│ Santa Ana Centro ┆ Z3 ┆ 7 │\n", + "│ Mercado Central ┆ Z1 ┆ 5 │\n", + "│ Plaza de la Cultura ┆ Z1 ┆ 5 │\n", + "│ UCR San Pedro ┆ Z1 ┆ 5 │\n", + "└───────────────────────────┴─────────┴─────────────┘\n" + ] + } + ], + "source": [ + "busiest = pl.read_parquet(PROCESSED / \"busiest_stops.parquet\")\n", + "print(busiest.head(10).select([\"stop_name\",\"zone_id\",\"trip_visits\"]))" + ] + }, + { + "cell_type": "markdown", + "id": "c9d0e1f2", + "metadata": {}, + "source": [ + "## 4. Frecuencia de servicio por franja horaria" + ] + }, + { + "cell_type": "code", + "execution_count": 13, + "id": "d0e1f2a3", + "metadata": {}, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "shape: (3, 2)\n", + "┌──────────┬─────────────┐\n", + "│ period ┆ total_trips │\n", + "│ --- ┆ --- │\n", + "│ str ┆ u32 │\n", + "╞══════════╪═════════════╡\n", + "│ mañana ┆ 14 │\n", + "│ mediodía ┆ 6 │\n", + "│ tarde ┆ 5 │\n", + "└──────────┴─────────────┘\n" + ] + } + ], + "source": [ + "freq = pl.read_parquet(PROCESSED / \"service_frequency.parquet\")\n", + "print(freq.group_by(\"period\").agg(pl.col(\"trip_count\").sum().alias(\"total_trips\")).sort(\"total_trips\", descending=True))" + ] + }, + { + "cell_type": "markdown", + "id": "e1f2a3b4", + "metadata": {}, + "source": [ + "## 5. Series temporales - velocidad de la flota con time_bucket()" + ] + }, + { + "cell_type": "code", + "execution_count": 15, + "id": "f2a3b4c5", + "metadata": {}, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "shape: (72, 3)\n", + "┌─────────────────────────┬────────────┬───────────────┐\n", + "│ bucket ┆ vehicle_id ┆ avg_speed_kmh │\n", + "│ --- ┆ --- ┆ --- │\n", + "│ datetime[μs, UTC] ┆ str ┆ decimal[38,1] │\n", + "╞═════════════════════════╪════════════╪═══════════════╡\n", + "│ 2026-03-20 02:30:00 UTC ┆ BUS-101 ┆ 34.5 │\n", + "│ 2026-03-20 02:30:00 UTC ┆ BUS-102 ┆ 38.8 │\n", + "│ 2026-03-20 02:30:00 UTC ┆ BUS-201 ┆ 43.9 │\n", + "│ 2026-03-20 02:30:00 UTC ┆ BUS-301 ┆ 4.8 │\n", + "│ 2026-03-20 02:30:00 UTC ┆ BUS-401 ┆ 18.4 │\n", + "│ … ┆ … ┆ … │\n", + "│ 2026-03-20 04:20:00 UTC ┆ BUS-102 ┆ 35.3 │\n", + "│ 2026-03-20 04:20:00 UTC ┆ BUS-201 ┆ 15.3 │\n", + "│ 2026-03-20 04:20:00 UTC ┆ BUS-301 ┆ 7.5 │\n", + "│ 2026-03-20 04:20:00 UTC ┆ BUS-401 ┆ 45.5 │\n", + "│ 2026-03-20 04:20:00 UTC ┆ BUS-501 ┆ 52.6 │\n", + "└─────────────────────────┴────────────┴───────────────┘\n" + ] + } + ], + "source": [ + "speed = query(\n", + " \"SELECT time_bucket('10 minutes', recorded_at) AS bucket, vehicle_id,\"\n", + " \" ROUND(AVG(speed_kmh)::NUMERIC, 1) AS avg_speed_kmh\"\n", + " \" FROM vehicle_events WHERE recorded_at > NOW() - INTERVAL '2 hours'\"\n", + " \" GROUP BY bucket, vehicle_id ORDER BY bucket, vehicle_id\"\n", + ")\n", + "if speed.is_empty():\n", + " print(\"Sin datos. Ejecutar flows/analyze.py primero.\")\n", + "else:\n", + " print(speed)" + ] + }, + { + "cell_type": "markdown", + "id": "a3b4c5d6", + "metadata": {}, + "source": [ + "## 6. Paradas con mas abordajes en la ultima hora" + ] + }, + { + "cell_type": "code", + "execution_count": 16, + "id": "b4c5d6e7", + "metadata": {}, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "shape: (10, 3)\n", + "┌───────────────────────────┬───────────┬────────────┐\n", + "│ stop_name ┆ boardings ┆ alightings │\n", + "│ --- ┆ --- ┆ --- │\n", + "│ str ┆ i64 ┆ i64 │\n", + "╞═══════════════════════════╪═══════════╪════════════╡\n", + "│ Parque La Sabana ┆ 73 ┆ 67 │\n", + "│ Mercado Central ┆ 59 ┆ 55 │\n", + "│ Sabana Norte ┆ 59 ┆ 49 │\n", + "│ Escazú Centro ┆ 53 ┆ 15 │\n", + "│ Guachipelín ┆ 51 ┆ 39 │\n", + "│ Estadio Nacional ┆ 47 ┆ 30 │\n", + "│ Hospital San Juan de Dios ┆ 44 ┆ 59 │\n", + "│ Pavas Centro ┆ 36 ┆ 57 │\n", + "│ Plaza de la Cultura ┆ 30 ┆ 25 │\n", + "│ Terminal 7-10 (San José) ┆ 27 ┆ 60 │\n", + "└───────────────────────────┴───────────┴────────────┘\n" + ] + } + ], + "source": [ + "ridership = query(\n", + " \"SELECT s.stop_name, SUM(sr.boardings) AS boardings, SUM(sr.alightings) AS alightings\"\n", + " \" FROM stop_ridership sr JOIN stops s USING (stop_id)\"\n", + " \" WHERE sr.recorded_at > NOW() - INTERVAL '1 hour'\"\n", + " \" GROUP BY s.stop_name ORDER BY boardings DESC LIMIT 10\"\n", + ")\n", + "if ridership.is_empty():\n", + " print(\"Sin datos. Ejecutar flows/analyze.py primero.\")\n", + "else:\n", + " print(ridership)" + ] + }, + { + "cell_type": "markdown", + "id": "c5d6e7f8", + "metadata": {}, + "source": [ + "## 7. Ultimo punto registrado por vehiculo" + ] + }, + { + "cell_type": "code", + "execution_count": 17, + "id": "d6e7f8a9", + "metadata": {}, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "shape: (6, 4)\n", + "┌────────────┬─────────────────────────────┬───────────────┬───────────────┐\n", + "│ vehicle_id ┆ recorded_at ┆ speed_kmh ┆ heading │\n", + "│ --- ┆ --- ┆ --- ┆ --- │\n", + "│ str ┆ datetime[μs, UTC] ┆ decimal[38,1] ┆ decimal[38,0] │\n", + "╞════════════╪═════════════════════════════╪═══════════════╪═══════════════╡\n", + "│ BUS-101 ┆ 2026-03-20 04:20:44.992 UTC ┆ 24.4 ┆ 223 │\n", + "│ BUS-102 ┆ 2026-03-20 04:20:44.992 UTC ┆ 35.3 ┆ 276 │\n", + "│ BUS-201 ┆ 2026-03-20 04:20:44.992 UTC ┆ 15.3 ┆ 25 │\n", + "│ BUS-301 ┆ 2026-03-20 04:20:44.992 UTC ┆ 7.5 ┆ 191 │\n", + "│ BUS-401 ┆ 2026-03-20 04:20:44.992 UTC ┆ 45.5 ┆ 233 │\n", + "│ BUS-501 ┆ 2026-03-20 04:20:44.992 UTC ┆ 52.6 ┆ 338 │\n", + "└────────────┴─────────────────────────────┴───────────────┴───────────────┘\n", + "Velocidad media de la flota: 30.1 km/h\n" + ] + } + ], + "source": [ + "latest = query(\n", + " \"SELECT DISTINCT ON (vehicle_id) vehicle_id, recorded_at,\"\n", + " \" ROUND(speed_kmh::NUMERIC, 1) AS speed_kmh, ROUND(heading::NUMERIC, 0) AS heading\"\n", + " \" FROM vehicle_events ORDER BY vehicle_id, recorded_at DESC\"\n", + ")\n", + "if latest.is_empty():\n", + " print(\"Sin datos. Ejecutar flows/analyze.py primero.\")\n", + "else:\n", + " print(latest)\n", + " print(f\"Velocidad media de la flota: {latest['speed_kmh'].mean():.1f} km/h\")" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "56206174-af37-492f-89ae-3d63365ac79b", + "metadata": {}, + "outputs": [], + "source": [] + } + ], + "metadata": { + "kernelspec": { + "display_name": "Python 3 (ipykernel)", + "language": "python", + "name": "python3" + }, + "language_info": { + "codemirror_mode": { + "name": "ipython", + "version": 3 + }, + "file_extension": ".py", + "mimetype": "text/x-python", + "name": "python", + "nbconvert_exporter": "python", + "pygments_lexer": "ipython3", + "version": "3.12.13" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +} diff --git a/lab-04-data-pipeline/notebooks/ml_demand.ipynb b/lab-04-data-pipeline/notebooks/ml_demand.ipynb new file mode 100644 index 0000000..cb58f11 --- /dev/null +++ b/lab-04-data-pipeline/notebooks/ml_demand.ipynb @@ -0,0 +1,196 @@ +{ + "nbformat": 4, + "nbformat_minor": 5, + "metadata": { + "kernelspec": {"display_name": "Python 3", "language": "python", "name": "python3"}, + "language_info": {"name": "python", "version": "3.12.0"} + }, + "cells": [ + { + "cell_type": "markdown", + "id": "a1b2c3d4", + "metadata": {}, + "source": [ + "# Lab 04 — Prediccion de demanda por parada\n", + "\n", + "Modelo simple de prediccion de abordajes por parada y franja horaria\n", + "usando los datos de la hypertable `stop_ridership` en TimescaleDB.\n", + "\n", + "**Objetivo**: dado el historial de abordajes por parada en los ultimos\n", + "30 minutos, predecir la carga esperada en los proximos 15 minutos.\n", + "\n", + "**Tecnica**: regresion lineal con features de ventana temporal (rolling mean).\n", + "No se requieren dependencias de ML externas — solo Polars y stdlib.\n", + "\n", + "**Pre-requisito**: haber ejecutado `flows/analyze.py` al menos una vez\n", + "para tener datos en `stop_ridership`." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "b2c3d4e5", + "metadata": {}, + "outputs": [], + "source": [ + "import os\n", + "from pathlib import Path\n", + "\n", + "import polars as pl\n", + "import psycopg2\n", + "import psycopg2.extras\n", + "\n", + "DB_URL = os.environ['DATABASE_URL']\n", + "\n", + "\n", + "def query(sql):\n", + " with psycopg2.connect(DB_URL) as conn:\n", + " with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:\n", + " cur.execute(sql)\n", + " return pl.DataFrame([dict(r) for r in cur.fetchall()])\n", + "\n", + "\n", + "print('Conexion OK')" + ] + }, + { + "cell_type": "markdown", + "id": "c3d4e5f6", + "metadata": {}, + "source": ["## 1. Cargar serie temporal de abordajes por parada"] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "d4e5f6a7", + "metadata": {}, + "outputs": [], + "source": [ + "raw = query(\n", + " \"SELECT time_bucket('15 minutes', recorded_at) AS bucket,\"\n", + " \" stop_id,\"\n", + " \" SUM(boardings) AS boardings\"\n", + " \" FROM stop_ridership\"\n", + " \" GROUP BY bucket, stop_id\"\n", + " \" ORDER BY stop_id, bucket\"\n", + ")\n", + "print(f\"Registros totales: {raw.shape[0]}\")\n", + "print(raw.head(10))" + ] + }, + { + "cell_type": "markdown", + "id": "e5f6a7b8", + "metadata": {}, + "source": ["## 2. Feature engineering: rolling mean de 2 periodos anteriores"] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "f6a7b8c9", + "metadata": {}, + "outputs": [], + "source": [ + "# Para cada parada, calcular la media movil de los 2 buckets anteriores\n", + "# como predictor del bucket siguiente.\n", + "features = (\n", + " raw\n", + " .sort(['stop_id', 'bucket'])\n", + " .with_columns(\n", + " pl.col('boardings')\n", + " .shift(1)\n", + " .over('stop_id')\n", + " .alias('lag_1'),\n", + " pl.col('boardings')\n", + " .shift(2)\n", + " .over('stop_id')\n", + " .alias('lag_2'),\n", + " )\n", + " .with_columns(\n", + " ((pl.col('lag_1') + pl.col('lag_2')) / 2)\n", + " .alias('rolling_mean_2')\n", + " )\n", + " .drop_nulls()\n", + ")\n", + "print(features.head(12))" + ] + }, + { + "cell_type": "markdown", + "id": "a7b8c9d0", + "metadata": {}, + "source": ["## 3. Prediccion con media movil (baseline)"] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "b8c9d0e1", + "metadata": {}, + "outputs": [], + "source": [ + "# El predictor baseline usa rolling_mean_2 como estimacion del siguiente valor.\n", + "# Evaluacion: MAE (Mean Absolute Error)\n", + "results = features.with_columns(\n", + " (pl.col('boardings') - pl.col('rolling_mean_2'))\n", + " .abs()\n", + " .alias('abs_error')\n", + ")\n", + "\n", + "mae = results['abs_error'].mean()\n", + "print(f\"MAE global (baseline rolling mean): {mae:.2f} abordajes\")\n", + "\n", + "# MAE por parada\n", + "mae_by_stop = (\n", + " results\n", + " .group_by('stop_id')\n", + " .agg(pl.col('abs_error').mean().round(2).alias('mae'))\n", + " .sort('mae', descending=True)\n", + ")\n", + "print(mae_by_stop)" + ] + }, + { + "cell_type": "markdown", + "id": "c9d0e1f2", + "metadata": {}, + "source": ["## 4. Prediccion para el proximo bucket (inference)"] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "d0e1f2a3", + "metadata": {}, + "outputs": [], + "source": [ + "# Ultimo valor conocido por parada => prediccion del siguiente bucket\n", + "last_known = (\n", + " features\n", + " .sort('bucket', descending=True)\n", + " .group_by('stop_id')\n", + " .agg(\n", + " pl.col('bucket').first().alias('last_bucket'),\n", + " pl.col('rolling_mean_2').first().round(1).alias('predicted_next'),\n", + " )\n", + " .sort('predicted_next', descending=True)\n", + ")\n", + "print('Prediccion de abordajes para el siguiente intervalo de 15 min:')\n", + "print(last_known)" + ] + }, + { + "cell_type": "markdown", + "id": "e1f2a3b4", + "metadata": {}, + "source": [ + "## Notas\n", + "\n", + "- Los datos de `stop_ridership` son **simulados** con valores aleatorios\n", + " uniformes, por lo que el MAE refleja ruido puro (sin patron real).\n", + "- Con datos reales de torniquetes o contadores de pasajeros, el modelo\n", + " captaria patrones de hora pico que reduciran el MAE significativamente.\n", + "- El siguiente paso natural seria un modelo de regresion con features\n", + " de hora del dia, dia de la semana y temperatura (via API meteorologica)." + ] + } + ] +} diff --git a/lab-04-data-pipeline/pyproject.toml b/lab-04-data-pipeline/pyproject.toml new file mode 100644 index 0000000..92119d1 --- /dev/null +++ b/lab-04-data-pipeline/pyproject.toml @@ -0,0 +1,19 @@ +[project] +name = "lab-04-data-pipeline" +version = "0.1.0" +requires-python = ">=3.12" +dependencies = [ + "prefect>=3.0", + "polars>=1.0", + "pyarrow>=16.0", + "psycopg2-binary>=2.9", + "python-dotenv>=1.0", + "pytest>=8.0", + "pytest-asyncio>=0.23", + "jupyterlab>=4.0", +] + +[tool.pytest.ini_options] +asyncio_mode = "auto" +pythonpath = ["."] +testpaths = ["tests"] diff --git a/lab-04-data-pipeline/tests/__init__.py b/lab-04-data-pipeline/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/lab-04-data-pipeline/tests/conftest.py b/lab-04-data-pipeline/tests/conftest.py new file mode 100644 index 0000000..524ddfd --- /dev/null +++ b/lab-04-data-pipeline/tests/conftest.py @@ -0,0 +1,73 @@ +""" +Fixtures compartidas para el Lab 04. + +Estrategia de aislamiento +-------------------------- +- El schema se aplica una sola vez por sesión (scope="session"). +- Antes de cada test se truncan todas las tablas de datos para garantizar + un estado limpio sin dependencias entre tests. +- Los flows de Prefect se ejecutan en modo local (sin servidor) estableciendo + PREFECT_API_URL vacío antes de importar cualquier módulo de Prefect. +""" + +import os +from pathlib import Path + +import psycopg2 +import pytest + +# Forzar ejecución local de Prefect sin servidor +os.environ.setdefault("PREFECT_API_URL", "") + +SCHEMA_FILE = Path(__file__).parent.parent / "models" / "timescale_schema.sql" + +# Tablas en orden inverso al de FK para poder truncar sin conflictos +TRUNCATE_ORDER = [ + "vehicle_events", + "stop_ridership", + "stop_times", + "trips", + "calendar", + "stops", + "routes", +] + + +@pytest.fixture(scope="session") +def db_url() -> str: + url = os.environ.get("DATABASE_URL") + if not url: + pytest.skip("DATABASE_URL no definida — " + "requiere TimescaleDB en Docker.") + return url + + +@pytest.fixture(scope="session", autouse=True) +def apply_schema(db_url: str) -> None: + """Crea tablas e hypertables una sola vez por sesión de tests.""" + sql = SCHEMA_FILE.read_text() + conn = psycopg2.connect(db_url) + conn.autocommit = True + with conn.cursor() as cur: + cur.execute(sql) + conn.close() + + +@pytest.fixture(autouse=True) +def clean_tables(db_url: str): + """Trunca todas las tablas antes de cada test.""" + yield + conn = psycopg2.connect(db_url) + with conn.cursor() as cur: + for table in TRUNCATE_ORDER: + cur.execute(f"TRUNCATE TABLE {table} CASCADE") # noqa: S608 + conn.commit() + conn.close() + + +@pytest.fixture +def db_conn(db_url: str): + """Conexión psycopg2 lista para usar en tests individuales.""" + conn = psycopg2.connect(db_url) + yield conn + conn.close() diff --git a/lab-04-data-pipeline/tests/test_flows.py b/lab-04-data-pipeline/tests/test_flows.py new file mode 100644 index 0000000..c5991b3 --- /dev/null +++ b/lab-04-data-pipeline/tests/test_flows.py @@ -0,0 +1,169 @@ +""" +Tests de integración para los tres flows del Lab 04. + +Todos los tests usan TimescaleDB real (sin mocks). +El fixture `clean_tables` en conftest garantiza aislamiento entre tests. +""" + +from pathlib import Path + +import polars as pl +import psycopg2.extras + +from flows.ingest_gtfs import ingest_gtfs +from flows.transform import transform_gtfs +from flows.analyze import analyze_timescale, VEHICLE_IDS + +OUTPUT_DIR = Path(__file__).parent.parent / "data" / "processed" + + +# ── Helpers ────────────────────────────────────────────────────────────────── + +def count_rows(conn, table: str) -> int: + with conn.cursor() as cur: + cur.execute(f"SELECT COUNT(*) FROM {table}") # noqa: S608 + return cur.fetchone()[0] + + +# ── ingest_gtfs ────────────────────────────────────────────────────────────── + +class TestIngestGtfs: + + def test_returns_counts_for_all_entities(self): + result = ingest_gtfs() + assert set(result.keys()) == {"stops", "routes", + "calendar", "trips", "stop_times"} + + def test_loads_correct_row_counts(self, db_conn): + ingest_gtfs() + assert count_rows(db_conn, "stops") == 23 + assert count_rows(db_conn, "routes") == 5 + assert count_rows(db_conn, "calendar") == 3 + assert count_rows(db_conn, "trips") == 25 + + def test_stop_times_loaded(self, db_conn): + result = ingest_gtfs() + assert result["stop_times"] > 0 + assert count_rows(db_conn, "stop_times") == result["stop_times"] + + def test_stops_have_valid_coordinates(self, db_conn): + ingest_gtfs() + with db_conn.cursor( + cursor_factory=psycopg2.extras.RealDictCursor) as cur: + cur.execute("SELECT stop_lat, stop_lon FROM stops") + rows = cur.fetchall() + for row in rows: + assert 8.0 < row["stop_lat"] < 12.0, "Latitud fuera del rango de Costa Rica" + assert -86.0 < row["stop_lon"] < -82.0, "Longitud fuera del rango de Costa Rica" + + def test_idempotent_second_run_does_not_duplicate(self, db_conn): + """Correr el flow dos veces no debe + duplicar registros (ON CONFLICT DO UPDATE).""" + ingest_gtfs() + ingest_gtfs() + assert count_rows(db_conn, "stops") == 23 + assert count_rows(db_conn, "routes") == 5 + assert count_rows(db_conn, "trips") == 25 + + def test_foreign_keys_are_consistent(self, db_conn): + """Todos los trip_id en stop_times deben existir en trips.""" + ingest_gtfs() + with db_conn.cursor() as cur: + cur.execute(""" + SELECT COUNT(*) FROM stop_times st + LEFT JOIN trips t USING (trip_id) + WHERE t.trip_id IS NULL + """) + orphaned = cur.fetchone()[0] + assert orphaned == 0, f"{orphaned} stop_times sin trip padre" + + +# ── transform_gtfs ─────────────────────────────────────────────────────────── + +class TestTransformGtfs: + + def setup_method(self): + """Ingesta previa necesaria para tener datos que transformar.""" + ingest_gtfs() + + def test_creates_parquet_files(self): + transform_gtfs() + for name in ("route_stats", "busiest_stops", "service_frequency"): + path = OUTPUT_DIR / f"{name}.parquet" + assert path.exists(), f"Parquet no creado: {path}" + assert path.stat().st_size > 0 + + def test_route_stats_covers_all_routes(self): + transform_gtfs() + df = pl.read_parquet(OUTPUT_DIR / "route_stats.parquet") + assert df.shape[0] == 5 # 5 rutas en el feed de ejemplo + assert "route_id" in df.columns + assert "total_trips" in df.columns + assert "avg_stops_trip" in df.columns + + def test_busiest_stops_sorted_descending(self): + transform_gtfs() + df = pl.read_parquet(OUTPUT_DIR / "busiest_stops.parquet") + visits = df["trip_visits"].to_list() + assert visits == sorted(visits, reverse=True), "busiest_stops no está ordenado" + + def test_busiest_stops_covers_all_stops(self): + transform_gtfs() + df = pl.read_parquet(OUTPUT_DIR / "busiest_stops.parquet") + assert df.shape[0] == 23 # 23 paradas en el feed de ejemplo + + def test_service_frequency_has_valid_periods(self): + transform_gtfs() + df = pl.read_parquet(OUTPUT_DIR / "service_frequency.parquet") + valid_periods = {"mañana", "mediodía", "tarde"} + actual_periods = set(df["period"].to_list()) + assert actual_periods.issubset(valid_periods) + + +# ── analyze_timescale ──────────────────────────────────────────────────────── + +class TestAnalyzeTimescale: + + def setup_method(self): + """Ingesta previa necesaria para tener stop_ids válidos.""" + ingest_gtfs() + + def test_seeds_vehicle_events(self, db_conn): + analyze_timescale() + count = count_rows(db_conn, "vehicle_events") + # 6 vehículos × 24 eventos = 144 eventos esperados + assert count == len(VEHICLE_IDS) * 24 + + def test_seeds_stop_ridership(self, db_conn): + analyze_timescale() + count = count_rows(db_conn, "stop_ridership") + assert count > 0 + + def test_returns_fleet_summary_keys(self): + result = analyze_timescale() + summary = result["fleet_summary"] + assert "active_vehicles" in summary + assert "avg_speed_kmh" in summary + assert "max_speed_kmh" in summary + + def test_fleet_summary_active_vehicles(self): + result = analyze_timescale() + active = result["fleet_summary"]["active_vehicles"] + assert 0 < active <= len(VEHICLE_IDS) + + def test_speed_buckets_returns_rows(self): + result = analyze_timescale() + assert result["speed_buckets_rows"] > 0 + + def test_vehicle_events_have_valid_coordinates(self, db_conn): + analyze_timescale() + with db_conn.cursor( + cursor_factory=psycopg2.extras.RealDictCursor) as cur: + cur.execute("SELECT latitude, " + "longitude FROM vehicle_events LIMIT 50") + rows = cur.fetchall() + for row in rows: + # Los eventos simulados parten del centro de San José + # con drift ±0.001° por paso × 24 pasos = ±0.024° máximo + assert 9.85 < row["latitude"] < 10.02 + assert -84.20 < row["longitude"] < -83.95