trainvel/sncf/api/views.py

302 lines
14 KiB
Python

from datetime import datetime, timedelta, date
from django.db.models import F, Q, Value, When, Case, Exists, OuterRef
from django.utils.decorators import method_decorator
from django.views.decorators.cache import cache_control
from django.views.decorators.http import last_modified
from django_filters.rest_framework import DjangoFilterBackend
from rest_framework import viewsets
from rest_framework.filters import OrderingFilter, SearchFilter
from sncf.api.serializers import AgencySerializer, StopSerializer, RouteSerializer, TripSerializer, \
StopTimeSerializer, CalendarSerializer, CalendarDateSerializer, TransferSerializer, \
FeedInfoSerializer, TripUpdateSerializer, StopTimeUpdateSerializer
from sncfgtfs.models import Agency, Stop, Route, Trip, StopTime, Calendar, CalendarDate, \
Transfer, FeedInfo, TripUpdate, StopTimeUpdate
CACHE_CONTROL = cache_control(max_age=7200)
LAST_MODIFIED = last_modified(lambda *args, **kwargs: datetime.fromisoformat(
FeedInfo.objects.get(publisher_name="SNCF_default").version))
LOOKUP_VALUE_REGEX = r"[\w.: |-]+"
@method_decorator(name='list', decorator=[CACHE_CONTROL, LAST_MODIFIED])
@method_decorator(name='retrieve', decorator=[CACHE_CONTROL, LAST_MODIFIED])
class AgencyViewSet(viewsets.ReadOnlyModelViewSet):
queryset = Agency.objects.all()
serializer_class = AgencySerializer
filter_backends = [DjangoFilterBackend]
filterset_fields = '__all__'
lookup_value_regex = LOOKUP_VALUE_REGEX
@method_decorator(name='list', decorator=[CACHE_CONTROL, LAST_MODIFIED])
@method_decorator(name='retrieve', decorator=[CACHE_CONTROL, LAST_MODIFIED])
class StopViewSet(viewsets.ReadOnlyModelViewSet):
queryset = Stop.objects.all()
serializer_class = StopSerializer
filter_backends = [DjangoFilterBackend, SearchFilter]
filterset_fields = '__all__'
search_fields = ['name',]
lookup_value_regex = LOOKUP_VALUE_REGEX
@method_decorator(name='list', decorator=[CACHE_CONTROL, LAST_MODIFIED])
@method_decorator(name='retrieve', decorator=[CACHE_CONTROL, LAST_MODIFIED])
class RouteViewSet(viewsets.ReadOnlyModelViewSet):
queryset = Route.objects.all()
serializer_class = RouteSerializer
filter_backends = [DjangoFilterBackend]
filterset_fields = '__all__'
lookup_value_regex = LOOKUP_VALUE_REGEX
@method_decorator(name='list', decorator=[CACHE_CONTROL, LAST_MODIFIED])
@method_decorator(name='retrieve', decorator=[CACHE_CONTROL, LAST_MODIFIED])
class TripViewSet(viewsets.ReadOnlyModelViewSet):
queryset = Trip.objects.all()
serializer_class = TripSerializer
filter_backends = [DjangoFilterBackend]
filterset_fields = '__all__'
lookup_value_regex = LOOKUP_VALUE_REGEX
@method_decorator(name='list', decorator=[CACHE_CONTROL, LAST_MODIFIED])
@method_decorator(name='retrieve', decorator=[CACHE_CONTROL, LAST_MODIFIED])
class StopTimeViewSet(viewsets.ReadOnlyModelViewSet):
queryset = StopTime.objects.order_by('id').all()
serializer_class = StopTimeSerializer
filter_backends = [DjangoFilterBackend, OrderingFilter]
filterset_fields = '__all__'
ordering_fields = ['arrival_time', 'departure_time', 'stop_sequence', ]
ordering = ['stop_sequence', ]
lookup_value_regex = LOOKUP_VALUE_REGEX
@method_decorator(name='list', decorator=[CACHE_CONTROL, LAST_MODIFIED])
@method_decorator(name='retrieve', decorator=[CACHE_CONTROL, LAST_MODIFIED])
class CalendarViewSet(viewsets.ReadOnlyModelViewSet):
queryset = Calendar.objects.all()
serializer_class = CalendarSerializer
filter_backends = [DjangoFilterBackend]
filterset_fields = '__all__'
lookup_value_regex = LOOKUP_VALUE_REGEX
@method_decorator(name='list', decorator=[CACHE_CONTROL, LAST_MODIFIED])
@method_decorator(name='retrieve', decorator=[CACHE_CONTROL, LAST_MODIFIED])
class CalendarDateViewSet(viewsets.ReadOnlyModelViewSet):
queryset = CalendarDate.objects.all()
serializer_class = CalendarDateSerializer
filter_backends = [DjangoFilterBackend]
filterset_fields = '__all__'
lookup_value_regex = LOOKUP_VALUE_REGEX
@method_decorator(name='list', decorator=[CACHE_CONTROL, LAST_MODIFIED])
@method_decorator(name='retrieve', decorator=[CACHE_CONTROL, LAST_MODIFIED])
class TransferViewSet(viewsets.ReadOnlyModelViewSet):
queryset = Transfer.objects.all()
serializer_class = TransferSerializer
filter_backends = [DjangoFilterBackend]
lookup_value_regex = LOOKUP_VALUE_REGEX
@method_decorator(name='list', decorator=[CACHE_CONTROL, LAST_MODIFIED])
@method_decorator(name='retrieve', decorator=[CACHE_CONTROL, LAST_MODIFIED])
class FeedInfoViewSet(viewsets.ReadOnlyModelViewSet):
queryset = FeedInfo.objects.all()
serializer_class = FeedInfoSerializer
filter_backends = [DjangoFilterBackend]
filterset_fields = '__all__'
lookup_value_regex = LOOKUP_VALUE_REGEX
class TripUpdateViewSet(viewsets.ReadOnlyModelViewSet):
queryset = TripUpdate.objects.all()
serializer_class = TripUpdateSerializer
filter_backends = [DjangoFilterBackend]
filterset_fields = '__all__'
lookup_value_regex = LOOKUP_VALUE_REGEX
class StopTimeUpdateViewSet(viewsets.ReadOnlyModelViewSet):
queryset = StopTimeUpdate.objects.all()
serializer_class = StopTimeUpdateSerializer
filter_backends = [DjangoFilterBackend]
filterset_fields = '__all__'
lookup_value_regex = LOOKUP_VALUE_REGEX
class NextDeparturesViewSet(viewsets.ReadOnlyModelViewSet):
queryset = StopTime.objects.none()
serializer_class = StopTimeSerializer
filter_backends = [DjangoFilterBackend]
def get_queryset(self):
now = datetime.now()
stop_id = self.request.query_params.get('stop_id', None)
stop_name = self.request.query_params.get('stop_name', None)
query_date = date.fromisoformat(self.request.query_params.get('date', now.date().isoformat()))
query_time = self.request.query_params.get('time', now.time().isoformat(timespec='seconds'))
query_time = timedelta(seconds=int(query_time[:2]) * 3600
+ int(query_time[3:5]) * 60
+ (int(query_time[6:]) if len(query_time) > 6 else 0))
yesterday = query_date - timedelta(days=1)
time_yesterday = query_time + timedelta(days=1)
tomorrow = query_date + timedelta(days=1)
stop_filter = Q(stop__location_type=0)
if stop_id:
stop = Stop.objects.get(id=stop_id)
stops = Stop.objects.filter(Q(id=stop_id)
| Q(parent_station=stop_id))
if stop.location_type == 0 and stop.parent_station_id is not None:
stops |= Stop.objects.filter(parent_station=stop.parent_station_id)
stop_filter = Q(stop__in=stops.values_list('id', flat=True))
elif stop_name:
stops = Stop.objects.filter(name__iexact=stop_name).values_list('id', flat=True)
stop_filter = Q(stop__in=stops)
def calendar_filter(d: date):
return Q(trip__service_id__in=CalendarDate.objects.filter(date=d, exception_type=1)
.values_list('service_id')) \
| Q(trip__service_id__in=Calendar.objects.filter(
start_date__lte=d,
end_date__gte=d,
**{f"{d:%A}".lower(): True})
.filter(~Q(id__in=CalendarDate.objects.filter(date=d, exception_type=2)
.values_list('service_id', flat=True)))
.values_list('id'))
def stop_time_update_qs(d: date):
return StopTimeUpdate.objects.filter(trip_update__start_date=d) \
.exclude(departure_time=datetime.fromtimestamp(0)).filter(stop_time_id=OuterRef('pk'))
def departure_time_real(d: date):
return Case(
When(
condition=Exists(stop_time_update_qs(d)),
then=F('departure_time') + stop_time_update_qs(d).values('departure_delay'),
),
default=F('departure_time'),
)
def canceled_filter(d: date):
return Exists(stop_time_update_qs(d).filter(Q(schedule_relationship=1) | Q(schedule_relationship=3)))
qs_today = StopTime.objects.filter(stop_filter) \
.annotate(departure_time_real=departure_time_real(query_date)) \
.filter(departure_time_real__gte=query_time) \
.filter(Q(pickup_type=0) | canceled_filter(query_date)) \
.filter(calendar_filter(query_date)) \
.annotate(departure_date=Value(query_date)) \
.annotate(departure_time_24h=F('departure_time'))
qs_yesterday = StopTime.objects.filter(stop_filter) \
.annotate(departure_time_real=departure_time_real(query_date)) \
.filter(departure_time_real__gte=time_yesterday) \
.filter(Q(pickup_type=0) | canceled_filter(yesterday)) \
.filter(calendar_filter(yesterday)) \
.annotate(departure_date=Value(yesterday)) \
.annotate(departure_time_24h=F('departure_time') - timedelta(days=1))
qs_tomorrow = StopTime.objects.filter(stop_filter) \
.annotate(departure_time_real=departure_time_real(query_date)) \
.filter(departure_time_real__gte=timedelta(0)) \
.filter(Q(pickup_type=0) | canceled_filter(tomorrow)) \
.filter(calendar_filter(tomorrow)) \
.annotate(departure_date=Value(tomorrow)) \
.annotate(departure_time_24h=F('departure_time') + timedelta(days=1))
return qs_today.union(qs_yesterday).union(qs_tomorrow).order_by("departure_time_24h").all()
class NextArrivalsViewSet(viewsets.ReadOnlyModelViewSet):
queryset = StopTime.objects.none()
serializer_class = StopTimeSerializer
filter_backends = [DjangoFilterBackend]
def get_queryset(self):
now = datetime.now()
stop_id = self.request.query_params.get('stop_id', None)
stop_name = self.request.query_params.get('stop_name', None)
query_date = date.fromisoformat(self.request.query_params.get('date', now.date().isoformat()))
query_time = self.request.query_params.get('time', now.time().isoformat(timespec='seconds'))
query_time = timedelta(seconds=int(query_time[:2]) * 3600
+ int(query_time[3:5]) * 60
+ (int(query_time[6:]) if len(query_time) > 6 else 0))
query_time -= timedelta(minutes=5) # Keep the last trains of the 5 previous minutes
yesterday = query_date - timedelta(days=1)
time_yesterday = query_time + timedelta(days=1)
tomorrow = query_date + timedelta(days=1)
stop_filter = Q(stop__location_type=0)
if stop_id:
stop = Stop.objects.get(id=stop_id)
stops = Stop.objects.filter(Q(id=stop_id)
| Q(parent_station=stop_id))
if stop.location_type == 0 and stop.parent_station_id is not None:
stops |= Stop.objects.filter(parent_station=stop.parent_station_id)
stop_filter = Q(stop__in=stops.values_list('id', flat=True))
elif stop_name:
stops = Stop.objects.filter(name__iexact=stop_name).values_list('id', flat=True)
stop_filter = Q(stop__in=stops)
def calendar_filter(d: date):
return Q(trip__service_id__in=CalendarDate.objects.filter(date=d, exception_type=1)
.values_list('service_id')) \
| Q(trip__service_id__in=Calendar.objects.filter(
start_date__lte=d,
end_date__gte=d,
**{f"{d:%A}".lower(): True})
.filter(~Q(id__in=CalendarDate.objects.filter(date=d, exception_type=2)
.values_list('service_id', flat=True)))
.values_list('id'))
def stop_time_update_qs(d: date):
return StopTimeUpdate.objects.filter(trip_update__start_date=d) \
.exclude(arrival_time=datetime.fromtimestamp(0)).filter(stop_time_id=OuterRef('pk'))
def arrival_time_real(d: date):
return Case(
When(
condition=Exists(stop_time_update_qs(d)),
then=F('arrival_time') + stop_time_update_qs(d).values('arrival_delay'),
),
default=F('arrival_time'),
)
def canceled_filter(d: date):
return Exists(stop_time_update_qs(d).filter(Q(schedule_relationship=1) | Q(schedule_relationship=3)))
qs_today = StopTime.objects.filter(stop_filter) \
.annotate(arrival_time_real=arrival_time_real(query_date)) \
.filter(arrival_time_real__gte=query_time) \
.filter(Q(drop_off_type=0) | canceled_filter(query_date)) \
.filter(calendar_filter(query_date)) \
.annotate(arrival_date=Value(query_date)) \
.annotate(arrival_time_24h=F('arrival_time'))
qs_yesterday = StopTime.objects.filter(stop_filter) \
.annotate(arrival_time_real=arrival_time_real(yesterday)) \
.filter(arrival_time_real__gte=time_yesterday) \
.filter(Q(drop_off_type=0) | canceled_filter(yesterday)) \
.filter(calendar_filter(yesterday)) \
.annotate(arrival_date=Value(yesterday)) \
.annotate(arrival_time_24h=F('arrival_time') - timedelta(days=1))
qs_tomorrow = StopTime.objects.filter(stop_filter) \
.annotate(arrival_time_real=arrival_time_real(tomorrow)) \
.filter(arrival_time_real__gte=timedelta(0)) \
.filter(Q(drop_off_type=0) | canceled_filter(tomorrow)) \
.filter(calendar_filter(tomorrow)) \
.annotate(arrival_date=Value(tomorrow)) \
.annotate(arrival_time_24h=F('arrival_time') + timedelta(days=1))
return qs_today.union(qs_yesterday).union(qs_tomorrow).order_by("arrival_time_24h").all()