Skip to content

Feature/asyncapi websocket spec #55

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
BrandonTrigueros wants to merge 10 commits into
base: main
Choose a base branch
from
Open

Feature/asyncapi websocket spec #55

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
68fab01
docs: recover AsyncAPI 3.0 specification for WebSocket channels
BrandonTrigueros Jan 19, 2026
04897c1
docs: recover AsyncAPI research documentation
BrandonTrigueros Jan 19, 2026
af4ae63
feat(asyncapi): add production features to WebSocket spec
BrandonTrigueros Jan 23, 2026
8b6673e
feat: Complete TURNO 2 - TripConsumer WebSocket implementation
BrandonTrigueros Jan 25, 2026
493270b
feat(websocket): Implement RouteConsumer with direction filtering and…
BrandonTrigueros Jan 25, 2026
1a1c0f6
feat(websocket): TURNO 4 - Celery broadcasting integration
BrandonTrigueros Jan 27, 2026
c49a808
feat(websocket): complete frontend integration with production-ready …
BrandonTrigueros Jan 27, 2026
5ce3934
fix(websocket): resolve JavaScript error and API query issues
BrandonTrigueros Jan 27, 2026
bc6c95a
fix(websocket): remove route validation to allow dynamic route_ids
BrandonTrigueros Jan 27, 2026
e972fb6
TURNO 4+5: WebSocket JSON sanitization for NaN/Infinity, frontend err…
BrandonTrigueros Jan 27, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion .env
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ CELERY_BROKER_URL=redis://redis:6379/0
CELERY_RESULT_BACKEND=redis://redis:6379/0

# Django Settings
ALLOWED_HOSTS=localhost,127.0.0.1,0.0.0.0
ALLOWED_HOSTS=localhost,127.0.0.1,0.0.0.0,172.26.163.149

# Static and Media Files
STATIC_URL=/static/
Expand Down
4 changes: 4 additions & 0 deletions api/urls.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,10 @@
path("next-trips/", views.NextTripView.as_view(), name="next-trips"),
path("next-stops/", views.NextStopView.as_view(), name="next-stops"),
path("route-stops/", views.RouteStopView.as_view(), name="route-stops"),
# Real-time tracking API endpoints
path('realtime/routes/active/', views.active_routes, name='api_active_routes'),
path('realtime/routes/<str:route_id>/', views.route_details, name='api_route_details'),
# Auth and docs
path("api-auth/", include("rest_framework.urls", namespace="rest_framework")),
path("docs/schema/", views.get_schema, name="schema"),
path("docs/", SpectacularRedocView.as_view(url_name="schema"), name="api_docs"),
Expand Down
170 changes: 170 additions & 0 deletions api/views.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
from datetime import datetime, timedelta
import pytz
from django.conf import settings
from django.utils import timezone

from .serializers import *

Expand Down Expand Up @@ -695,3 +696,172 @@ def get_calendar(date, current_feed):
service_id = calendar.service_id

return service_id


# ==========================================
# Real-time Tracking API Endpoints
# ==========================================

from rest_framework.decorators import api_view
from rest_framework.response import Response
from django.db.models import Count, Q
from datetime import timedelta


@api_view(['GET'])
def active_routes(request):
"""
Get list of routes with active vehicles.

Returns routes that have had vehicle updates in the last 10 minutes,
sorted by number of active vehicles descending.

Query Parameters:
- min_vehicles (int): Minimum number of vehicles (default: 1)
- limit (int): Maximum routes to return (default: 20)

Response:
{
"count": 5,
"routes": [
{
"route_id": "Red",
"route_short_name": "Red Line",
"route_long_name": "Red Line",
"route_type": 1,
"vehicle_count": 10,
"last_update": "2026-01-27T01:45:23Z"
},
...
]
}
"""
from gtfs.models import VehiclePosition

# Get query parameters
min_vehicles = int(request.GET.get('min_vehicles', 1))
limit = int(request.GET.get('limit', 20))

# Find routes with recent vehicle positions (last 10 minutes)
cutoff = timezone.now() - timedelta(minutes=10)

# Get route_ids with recent vehicles from VehiclePosition
# Note: VehiclePosition doesn't have ForeignKey to Route, only vehicle_trip_route_id CharField
recent_vehicles = VehiclePosition.objects.filter(
vehicle_timestamp__gte=cutoff,
vehicle_trip_route_id__isnull=False
).values('vehicle_trip_route_id').annotate(
vehicle_count=Count('id')
).filter(
vehicle_count__gte=min_vehicles
).order_by('-vehicle_count')[:limit]

# Build response with Route details
routes_data = []
for item in recent_vehicles:
route_id = item['vehicle_trip_route_id']
vehicle_count = item['vehicle_count']

# Try to get Route object for metadata (use first() to avoid MultipleObjectsReturned)
route = Route.objects.filter(route_id=route_id).first()

if route:
route_short_name = route.route_short_name or route_id
route_long_name = route.route_long_name or route_id
route_type = route.route_type
else:
# Route not in database, use route_id as fallback
route_short_name = route_id
route_long_name = route_id
route_type = 3 # Default to bus

# Get last update time
last_position = VehiclePosition.objects.filter(
vehicle_trip_route_id=route_id,
vehicle_timestamp__gte=cutoff
).order_by('-vehicle_timestamp').first()

routes_data.append({
'route_id': route_id,
'route_short_name': route_short_name,
'route_long_name': route_long_name,
'route_type': route_type,
'vehicle_count': vehicle_count,
'last_update': last_position.vehicle_timestamp.isoformat() if last_position else None
})

return Response({
'count': len(routes_data),
'routes': routes_data
})


@api_view(['GET'])
def route_details(request, route_id):
"""
Get details for a specific route.

Parameters:
- route_id: Route identifier

Response:
{
"route_id": "Red",
"route_short_name": "Red Line",
"route_long_name": "Red Line",
"route_type": 1,
"route_type_name": "Subway/Metro",
"vehicle_count": 10,
"directions": [
{"direction_id": 0, "vehicle_count": 5},
{"direction_id": 1, "vehicle_count": 5}
]
}
"""
from gtfs.models import VehiclePosition

try:
route = Route.objects.get(route_id=route_id)
except Route.DoesNotExist:
return Response({'error': 'Route not found'}, status=404)

# Route type names
route_type_names = {
0: "Tram/Light Rail",
1: "Subway/Metro",
2: "Rail",
3: "Bus",
4: "Ferry",
5: "Cable Car",
6: "Gondola",
7: "Funicular"
}

# Count vehicles by direction (last 10 minutes)
cutoff = timezone.now() - timedelta(minutes=10)
recent_vehicles = VehiclePosition.objects.filter(
vehicle_trip_route_id=route_id,
vehicle_timestamp__gte=cutoff
)

total_count = recent_vehicles.count()

# Count by direction
directions = []
for direction_id in [0, 1]:
count = recent_vehicles.filter(vehicle_trip_direction_id=direction_id).count()
if count > 0:
directions.append({
'direction_id': direction_id,
'vehicle_count': count
})

return Response({
'route_id': route.route_id,
'route_short_name': route.route_short_name or route.route_id,
'route_long_name': route.route_long_name or route.route_id,
'route_type': route.route_type,
'route_type_name': route_type_names.get(route.route_type, 'Unknown'),
'vehicle_count': total_count,
'directions': directions
})
8 changes: 6 additions & 2 deletions datahub/asgi.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,13 +12,17 @@
from channels.routing import ProtocolTypeRouter, URLRouter
from django.core.asgi import get_asgi_application

from feed.routing import websocket_urlpatterns
from feed.routing import websocket_urlpatterns as feed_ws_patterns
from websocket.routing import websocket_urlpatterns as websocket_ws_patterns

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "datahub.settings")

# Combine WebSocket URL patterns from both apps
all_websocket_urlpatterns = feed_ws_patterns + websocket_ws_patterns

application = ProtocolTypeRouter(
{
"http": get_asgi_application(),
"websocket": URLRouter(websocket_urlpatterns),
"websocket": URLRouter(all_websocket_urlpatterns),
}
)
13 changes: 13 additions & 0 deletions datahub/settings.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@
"feed.apps.FeedConfig",
"alerts.apps.AlertsConfig",
"api.apps.ApiConfig",
"websocket",
"rest_framework",
"rest_framework.authtoken",
"drf_spectacular",
Expand Down Expand Up @@ -138,6 +139,18 @@
CELERY_CACHE_BACKEND = "django-cache"
CELERY_RESULTS_EXTENDED = True

# Celery Beat schedule for periodic tasks
CELERY_BEAT_SCHEDULE = {
"get-vehicle-positions-every-30-seconds": {
"task": "feed.tasks.get_vehicle_positions",
"schedule": 30.0, # Run every 30 seconds
},
"get-trip-updates-every-30-seconds": {
"task": "feed.tasks.get_trip_updates",
"schedule": 30.0, # Run every 30 seconds
},
}

# REST Framework settings

REST_FRAMEWORK = {
Expand Down
1 change: 1 addition & 0 deletions datahub/urls.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,4 +31,5 @@ def health_check(request):
path("gtfs/", include("gtfs.urls")),
path("status/", include("feed.urls")),
path("alertas/", include("alerts.urls")),
path("websocket/", include("websocket.urls")),
]
79 changes: 79 additions & 0 deletions demos/mbta/README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,79 @@
# MBTA Boston Demo Setup

This directory contains **demo-specific** setup scripts for Massachusetts Bay Transportation Authority (MBTA) data. These scripts are **NOT part of the production codebase** and serve only as examples for testing.

## Purpose

The Infobús real-time tracking system is **provider-agnostic**. This demo shows how to integrate with MBTA's GTFS-Realtime feeds as a proof-of-concept.

## Demo Scripts

### 1. `create_mbta_routes.py`
Creates MBTA routes in the database directly.

**Usage:**
```bash
docker compose exec celery-worker bash -c "cd /app && uv run python demos/mbta/create_mbta_routes.py"
```

**What it does:**
- Creates MBTA feed and agency
- Adds 10 common routes (Red, Blue, Orange, Green lines, buses)
- Only for initial demo setup

### 2. `load_mbta_routes.py`
Management command to load routes from MBTA's real-time protobuf feed.

**Usage:**
```bash
docker compose exec web uv run python manage.py load_mbta_routes
```

**What it does:**
- Fetches live VehiclePositions.pb from MBTA
- Extracts unique route IDs
- Creates Route objects automatically

## Integration with Production Code

The main Infobús system expects:
1. A configured `GTFSProvider` with URLs
2. Routes in the database (from GTFS Schedule or manual creation)
3. Celery tasks configured to fetch from provider URLs

For **production deployment**:
1. Configure your provider in admin panel
2. Import GTFS Schedule data (not covered by demo)
3. Configure Celery tasks with provider URLs
4. Deploy web client pointing to `/realtime/` endpoint

## Demo Configuration

To use MBTA demo data, set in your environment:

```python
# settings.py or environment variable
REALTIME_DEMO_PROVIDER = 'MBTA'
REALTIME_DEMO_CENTER = [42.3601, -71.0589] # Boston coordinates
```

## Removing Demo Data

To clean up MBTA demo data:
```bash
docker compose exec web uv run python manage.py shell -c "
from gtfs.models import Route, Feed, Agency
Feed.objects.filter(feed_id__in=['mbta', 'MBTA_FEED']).delete()
"
```

## Adapting for Other Providers

1. Copy this directory: `cp -r demos/mbta demos/your_provider`
2. Update route metadata in scripts
3. Change URLs to your provider's feeds
4. Update README with provider-specific instructions

---

**Remember:** This is a **demo setup**, not production code. Production systems should load GTFS Schedule data through proper ETL pipelines.
Loading