ynerant
/
trainvel
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add import script

This commit is contained in:
Emmy D'Anello 2024-01-27 10:43:59 +01:00
parent 90cf8c61a5
commit de6e231639
Signed by: ynerant
GPG Key ID: 3A75C55819C8CF85
3 changed files with 597 additions and 99 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

345
sncfgtfs/management/commands/update_sncf_gtfs.py Normal file
View File

@ -0,0 +1,345 @@
import csv
from datetime import datetime
from io import BytesIO
from zipfile import ZipFile
import requests
from django.core.management import BaseCommand
from sncfgtfs.models import Agency, Calendar, CalendarDate, FeedInfo, Route, Stop, StopTime, Transfer, Trip
class Command(BaseCommand):
help = "Update the SNCF GTFS database."
GTFS_FEEDS = {
"TGV": "https://eu.ftp.opendatasoft.com/sncf/gtfs/export_gtfs_voyages.zip",
"IC": "https://eu.ftp.opendatasoft.com/sncf/gtfs/export-intercites-gtfs-last.zip",
"TER": "https://eu.ftp.opendatasoft.com/sncf/gtfs/export-ter-gtfs-last.zip",
"TN": "https://eu.ftp.opendatasoft.com/sncf/gtfs/transilien-gtfs.zip",
}
def add_arguments(self, parser):
parser.add_argument('--bulk_size', type=int, default=1000, help='Number of objects to create in bulk.')
def handle(self, *args, **options):
bulk_size = options['bulk_size']
if not FeedInfo.objects.exists():
last_update_date = "1970-01-01"
else:
last_update_date = FeedInfo.objects.get().feed_version
for url in self.GTFS_FEEDS.values():
last_modified = requests.head(url).headers["Last-Modified"]
last_modified = datetime.strptime(last_modified, "%a, %d %b %Y %H:%M:%S %Z")
if last_modified.date().isoformat() > last_update_date:
break
else:
self.stdout.write(self.style.WARNING("Database already up-to-date."))
return
self.stdout.write("Updating database...")
for transport_type, feed_url in self.GTFS_FEEDS.items():
self.stdout.write(f"Downloading {transport_type} GTFS feed...")
with ZipFile(BytesIO(requests.get(feed_url).content)) as zipfile:
agencies = []
for agency_dict in csv.DictReader(zipfile.read("agency.txt").decode().splitlines()):
agency_dict: dict
agency = Agency(
id=agency_dict['agency_id'],
name=agency_dict['agency_name'],
url=agency_dict['agency_url'],
timezone=agency_dict['agency_timezone'],
lang=agency_dict['agency_lang'],
phone=agency_dict.get('agency_phone', ""),
email=agency_dict.get('agency_email', ""),
)
agencies.append(agency)
if agencies:
Agency.objects.bulk_create(agencies,
update_conflicts=True,
update_fields=['name', 'url', 'timezone', 'lang', 'phone', 'email'],
unique_fields=['id'])
agencies.clear()
stops = []
for stop_dict in csv.DictReader(zipfile.read("stops.txt").decode().splitlines()):
stop_dict: dict
stop = Stop(
id=stop_dict["stop_id"],
name=stop_dict['stop_name'],
desc=stop_dict['stop_desc'],
lat=stop_dict['stop_lat'],
lon=stop_dict['stop_lon'],
zone_id=stop_dict['zone_id'],
url=stop_dict['stop_url'],
location_type=stop_dict['location_type'],
parent_station_id=stop_dict['parent_station'] or None
if last_update_date != "1970-01-01" or transport_type != "TN" else None,
timezone=stop_dict.get('stop_timezone', ""),
wheelchair_boarding=stop_dict.get('wheelchair_boarding', 0),
level_id=stop_dict.get('level_id', ""),
platform_code=stop_dict.get('platform_code', ""),
)
stops.append(stop)
if len(stops) >= bulk_size:
Stop.objects.bulk_create(stops,
update_conflicts=True,
update_fields=['name', 'desc', 'lat', 'lon', 'zone_id', 'url',
'location_type', 'parent_station_id', 'timezone',
'wheelchair_boarding', 'level_id', 'platform_code'],
unique_fields=['id'])
stops.clear()
if stops:
Stop.objects.bulk_create(stops,
update_conflicts=True,
update_fields=['name', 'desc', 'lat', 'lon', 'zone_id', 'url',
'location_type', 'parent_station_id', 'timezone',
'wheelchair_boarding', 'level_id', 'platform_code'],
unique_fields=['id'])
stops.clear()
routes = []
for route_dict in csv.DictReader(zipfile.read("routes.txt").decode().splitlines()):
route_dict: dict
route = Route(
id=route_dict['route_id'],
agency_id=route_dict['agency_id'],
short_name=route_dict['route_short_name'],
long_name=route_dict['route_long_name'],
desc=route_dict['route_desc'],
type=route_dict['route_type'],
url=route_dict['route_url'],
color=route_dict['route_color'],
text_color=route_dict['route_text_color'],
)
routes.append(route)
if len(routes) >= bulk_size:
Route.objects.bulk_create(routes,
update_conflicts=True,
update_fields=['agency_id', 'short_name', 'long_name', 'desc',
'type', 'url', 'color', 'text_color'],
unique_fields=['id'])
routes.clear()
if routes:
Route.objects.bulk_create(routes,
update_conflicts=True,
update_fields=['agency_id', 'short_name', 'long_name', 'desc',
'type', 'url', 'color', 'text_color'],
unique_fields=['id'])
routes.clear()
calendar_ids = []
if "calendar.txt" in zipfile.namelist():
calendars = []
for calendar_dict in csv.DictReader(zipfile.read("calendar.txt").decode().splitlines()):
calendar_dict: dict
calendar = Calendar(
id=f"{transport_type}-{calendar_dict['service_id']}",
monday=calendar_dict['monday'],
tuesday=calendar_dict['tuesday'],
wednesday=calendar_dict['wednesday'],
thursday=calendar_dict['thursday'],
friday=calendar_dict['friday'],
saturday=calendar_dict['saturday'],
sunday=calendar_dict['sunday'],
start_date=calendar_dict['start_date'],
end_date=calendar_dict['end_date'],
transport_type=transport_type,
)
calendars.append(calendar)
calendar_ids.append(calendar.id)
if len(calendars) >= bulk_size:
Calendar.objects.bulk_create(calendars,
update_conflicts=True,
update_fields=['monday', 'tuesday', 'wednesday', 'thursday',
'friday', 'saturday', 'sunday', 'start_date',
'end_date', 'transport_type'],
unique_fields=['id'])
calendars.clear()
if calendars:
Calendar.objects.bulk_create(calendars, update_conflicts=True,
update_fields=['monday', 'tuesday', 'wednesday', 'thursday',
'friday', 'saturday', 'sunday', 'start_date',
'end_date', 'transport_type'],
unique_fields=['id'])
calendars.clear()
calendars = []
calendar_dates = []
for calendar_date_dict in csv.DictReader(zipfile.read("calendar_dates.txt").decode().splitlines()):
calendar_date_dict: dict
calendar_date = CalendarDate(
id=f"{transport_type}-{calendar_date_dict['service_id']}-{calendar_date_dict['date']}",
service_id=f"{transport_type}-{calendar_date_dict['service_id']}",
date=calendar_date_dict['date'],
exception_type=calendar_date_dict['exception_type'],
)
calendar_dates.append(calendar_date)
if calendar_date.service_id not in calendar_ids:
calendar = Calendar(
id=f"{transport_type}-{calendar_date_dict['service_id']}",
monday=False,
tuesday=False,
wednesday=False,
thursday=False,
friday=False,
saturday=False,
sunday=False,
start_date=calendar_date_dict['date'],
end_date=calendar_date_dict['date'],
transport_type=transport_type,
)
calendars.append(calendar)
if len(calendar_dates) >= bulk_size:
Calendar.objects.bulk_create(calendars,
update_conflicts=True,
update_fields=['end_date'],
unique_fields=['id'])
CalendarDate.objects.bulk_create(calendar_dates,
update_conflicts=True,
update_fields=['service_id', 'date', 'exception_type'],
unique_fields=['id'])
calendars.clear()
calendar_dates.clear()
if calendar_dates:
Calendar.objects.bulk_create(calendars,
update_conflicts=True,
update_fields=['end_date'],
unique_fields=['id'])
CalendarDate.objects.bulk_create(calendar_dates,
update_conflicts=True,
update_fields=['service_id', 'date', 'exception_type'],
unique_fields=['id'])
calendars.clear()
calendar_dates.clear()
trips = []
for trip_dict in csv.DictReader(zipfile.read("trips.txt").decode().splitlines()):
trip_dict: dict
trip = Trip(
id=trip_dict['trip_id'],
route_id=trip_dict['route_id'],
service_id=f"{transport_type}-{trip_dict['service_id']}",
headsign=trip_dict['trip_headsign'],
short_name=trip_dict.get('trip_short_name', ""),
direction_id=trip_dict['direction_id'] or None,
block_id=trip_dict['block_id'],
shape_id=trip_dict['shape_id'],
wheelchair_accessible=trip_dict.get('wheelchair_accessible', None),
bikes_allowed=trip_dict.get('bikes_allowed', None),
)
trips.append(trip)
if len(trips) >= bulk_size:
Trip.objects.bulk_create(trips,
update_conflicts=True,
update_fields=['route_id', 'service_id', 'headsign', 'short_name',
'direction_id', 'block_id', 'shape_id',
'wheelchair_accessible', 'bikes_allowed'],
unique_fields=['id'])
trips.clear()
if trips:
Trip.objects.bulk_create(trips,
update_conflicts=True,
update_fields=['route_id', 'service_id', 'headsign', 'short_name',
'direction_id', 'block_id', 'shape_id',
'wheelchair_accessible', 'bikes_allowed'],
unique_fields=['id'])
trips.clear()
stop_times = []
for stop_time_dict in csv.DictReader(zipfile.read("stop_times.txt").decode().splitlines()):
stop_time_dict: dict
arrival_next_day = False
arrival_time = stop_time_dict['arrival_time']
if int(arrival_time.split(":")[0]) >= 24:
split = arrival_time.split(':')
arrival_time = f"{int(split[0]) - 24:02}:{split[1]}:{split[2]}"
arrival_next_day = True
departure_time = stop_time_dict['departure_time']
if int(departure_time.split(":")[0]) >= 24:
split = departure_time.split(':')
departure_time = f"{int(split[0]) - 24:02}:{split[1]}:{split[2]}"
arrival_next_day = True
st = StopTime(
id=f"{stop_time_dict['trip_id']}-{stop_time_dict['stop_sequence']}",
trip_id=stop_time_dict['trip_id'],
arrival_time=arrival_time,
departure_time=departure_time,
arrival_next_day=arrival_next_day,
stop_id=stop_time_dict['stop_id'],
stop_sequence=stop_time_dict['stop_sequence'],
stop_headsign=stop_time_dict['stop_headsign'],
pickup_type=stop_time_dict['pickup_type'],
drop_off_type=stop_time_dict['drop_off_type'],
timepoint=stop_time_dict.get('timepoint', None),
)
stop_times.append(st)
if len(stop_times) >= bulk_size:
StopTime.objects.bulk_create(stop_times,
update_conflicts=True,
update_fields=['stop_id', 'arrival_time', 'departure_time',
'arrival_next_day', 'stop_headsign', 'pickup_type',
'drop_off_type', 'timepoint'],
unique_fields=['id'])
stop_times.clear()
if stop_times:
StopTime.objects.bulk_create(stop_times,
update_conflicts=True,
update_fields=['stop_id', 'arrival_time', 'departure_time',
'arrival_next_day', 'stop_headsign', 'pickup_type',
'drop_off_type', 'timepoint'],
unique_fields=['id'])
stop_times.clear()
transfers = []
for transfer_dict in csv.DictReader(zipfile.read("transfers.txt").decode().splitlines()):
transfer_dict: dict
transfer = Transfer(
id=f"{transfer_dict['from_stop_id']}-{transfer_dict['to_stop_id']}",
from_stop_id=transfer_dict['from_stop_id'],
to_stop_id=transfer_dict['to_stop_id'],
transfer_type=transfer_dict['transfer_type'],
min_transfer_time=transfer_dict['min_transfer_time'],
)
transfers.append(transfer)
if len(transfers) >= bulk_size:
Transfer.objects.bulk_create(transfers,
update_conflicts=True,
update_fields=['transfer_type', 'min_transfer_time'],
unique_fields=['id'])
transfers.clear()
if transfers:
Transfer.objects.bulk_create(transfers,
update_conflicts=True,
update_fields=['transfer_type', 'min_transfer_time'],
unique_fields=['id'])
transfers.clear()
if "feed_info.txt" in zipfile.namelist():
for feed_info_dict in csv.DictReader(zipfile.read("feed_info.txt").decode().splitlines()):
feed_info_dict: dict
FeedInfo.objects.update_or_create(
feed_publisher_name=feed_info_dict['feed_publisher_name'],
defaults={
'feed_publisher_url': feed_info_dict['feed_publisher_url'],
'feed_lang': feed_info_dict['feed_lang'],
'feed_start_date': feed_info_dict['feed_start_date'],
'feed_end_date': feed_info_dict['feed_end_date'],
'feed_version': feed_info_dict['feed_version'],
}
)

173
sncfgtfs/migrations/0001_initial.py
View File

@ -1,4 +1,4 @@
# Generated by Django 5.0.1 on 2024-01-26 20:15 # Generated by Django 5.0.1 on 2024-01-27 09:09
import django.db.models.deletion import django.db.models.deletion
from django.db import migrations, models from django.db import migrations, models
@ -14,7 +14,7 @@ class Migration(migrations.Migration):
name="Agency", name="Agency",
fields=[ fields=[
( (
"agency_id", "id",
models.CharField( models.CharField(
max_length=255, max_length=255,
primary_key=True, primary_key=True,
@ -23,22 +23,34 @@ class Migration(migrations.Migration):
), ),
), ),
( (
"agency_name", "name",
models.CharField( models.CharField(
max_length=255, unique=True, verbose_name="Agency name" max_length=255, unique=True, verbose_name="Agency name"
), ),
), ),
("agency_url", models.URLField(verbose_name="Agency URL")), ("url", models.URLField(verbose_name="Agency URL")),
( (
"agency_timezone", "timezone",
models.CharField(max_length=255, verbose_name="Agency timezone"), models.CharField(max_length=255, verbose_name="Agency timezone"),
), ),
( (
"agency_lang", "lang",
models.CharField( models.CharField(
blank=True, max_length=255, verbose_name="Agency language" blank=True, max_length=255, verbose_name="Agency language"
), ),
), ),
(
"phone",
models.CharField(
blank=True, max_length=255, verbose_name="Agency phone"
),
),
(
"email",
models.EmailField(
blank=True, max_length=254, verbose_name="Agency email"
),
),
], ],
options={ options={
"verbose_name": "Agency", "verbose_name": "Agency",
@ -49,7 +61,7 @@ class Migration(migrations.Migration):
name="Calendar", name="Calendar",
fields=[ fields=[
( (
"service_id", "id",
models.CharField( models.CharField(
max_length=255, max_length=255,
primary_key=True, primary_key=True,
@ -66,6 +78,19 @@ class Migration(migrations.Migration):
("sunday", models.BooleanField(verbose_name="Sunday")), ("sunday", models.BooleanField(verbose_name="Sunday")),
("start_date", models.DateField(verbose_name="Start date")), ("start_date", models.DateField(verbose_name="Start date")),
("end_date", models.DateField(verbose_name="End date")), ("end_date", models.DateField(verbose_name="End date")),
(
"transport_type",
models.CharField(
choices=[
("TGV", "TGV"),
("TER", "TER"),
("IC", "Intercités"),
("TN", "Transilien"),
],
max_length=255,
verbose_name="Transport type",
),
),
], ],
options={ options={
"verbose_name": "Calendar", "verbose_name": "Calendar",
@ -76,9 +101,12 @@ class Migration(migrations.Migration):
name="FeedInfo", name="FeedInfo",
fields=[ fields=[
( (
"feed_id", "id",
models.SmallIntegerField( models.BigAutoField(
primary_key=True, serialize=False, verbose_name="Feed ID" auto_created=True,
primary_key=True,
serialize=False,
verbose_name="ID",
), ),
), ),
( (
@ -112,21 +140,41 @@ class Migration(migrations.Migration):
fields=[ fields=[
( (
"id", "id",
models.BigAutoField( models.CharField(
auto_created=True, max_length=255,
primary_key=True, primary_key=True,
serialize=False, serialize=False,
verbose_name="ID", verbose_name="ID",
), ),
), ),
("date", models.DateField(verbose_name="Date")), ("date", models.DateField(verbose_name="Date")),
("exception_type", models.IntegerField(verbose_name="Exception type")),
( (
"service_id", "exception_type",
models.IntegerField(
choices=[(1, "Added"), (2, "Removed")],
verbose_name="Exception type",
),
),
(
"transport_type",
models.CharField(
choices=[
("TGV", "TGV"),
("TER", "TER"),
("IC", "Intercités"),
("TN", "Transilien"),
],
max_length=255,
verbose_name="Transport type",
),
),
(
"service",
models.ForeignKey( models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE, on_delete=django.db.models.deletion.CASCADE,
related_name="dates",
to="sncfgtfs.calendar", to="sncfgtfs.calendar",
verbose_name="Service ID", verbose_name="Service",
), ),
), ),
], ],
@ -139,30 +187,30 @@ class Migration(migrations.Migration):
name="Route", name="Route",
fields=[ fields=[
( (
"route_id", "id",
models.CharField( models.CharField(
max_length=255, max_length=255,
primary_key=True, primary_key=True,
serialize=False, serialize=False,
verbose_name="Route ID", verbose_name="ID",
), ),
), ),
( (
"route_short_name", "short_name",
models.CharField(max_length=255, verbose_name="Route short name"), models.CharField(max_length=255, verbose_name="Route short name"),
), ),
( (
"route_long_name", "long_name",
models.CharField(max_length=255, verbose_name="Route long name"), models.CharField(max_length=255, verbose_name="Route long name"),
), ),
( (
"route_desc", "desc",
models.CharField( models.CharField(
blank=True, max_length=255, verbose_name="Route description" blank=True, max_length=255, verbose_name="Route description"
), ),
), ),
( (
"route_type", "type",
models.IntegerField( models.IntegerField(
choices=[ choices=[
(0, "Tram"), (0, "Tram"),
@ -177,15 +225,15 @@ class Migration(migrations.Migration):
verbose_name="Route type", verbose_name="Route type",
), ),
), ),
("route_url", models.URLField(blank=True, verbose_name="Route URL")), ("url", models.URLField(blank=True, verbose_name="Route URL")),
( (
"route_color", "color",
models.CharField( models.CharField(
blank=True, max_length=255, verbose_name="Route color" blank=True, max_length=255, verbose_name="Route color"
), ),
), ),
( (
"route_text_color", "text_color",
models.CharField( models.CharField(
blank=True, max_length=255, verbose_name="Route text color" blank=True, max_length=255, verbose_name="Route text color"
), ),
@ -194,6 +242,7 @@ class Migration(migrations.Migration):
"agency", "agency",
models.ForeignKey( models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE, on_delete=django.db.models.deletion.CASCADE,
related_name="routes",
to="sncfgtfs.agency", to="sncfgtfs.agency",
verbose_name="Agency ID", verbose_name="Agency ID",
), ),
@ -208,7 +257,7 @@ class Migration(migrations.Migration):
name="Stop", name="Stop",
fields=[ fields=[
( (
"stop_id", "id",
models.CharField( models.CharField(
max_length=255, max_length=255,
primary_key=True, primary_key=True,
@ -217,25 +266,22 @@ class Migration(migrations.Migration):
), ),
), ),
( (
"stop_code", "code",
models.CharField( models.CharField(
blank=True, max_length=255, verbose_name="Stop code" blank=True, max_length=255, verbose_name="Stop code"
), ),
), ),
("name", models.CharField(max_length=255, verbose_name="Stop name")),
( (
"stop_name", "desc",
models.CharField(max_length=255, verbose_name="Stop name"),
),
(
"stop_desc",
models.CharField( models.CharField(
blank=True, max_length=255, verbose_name="Stop description" blank=True, max_length=255, verbose_name="Stop description"
), ),
), ),
("stop_lon", models.FloatField(verbose_name="Stop longitude")), ("lon", models.FloatField(verbose_name="Stop longitude")),
("stop_lat", models.FloatField(verbose_name="Stop latitude")), ("lat", models.FloatField(verbose_name="Stop latitude")),
("zone_id", models.CharField(max_length=255, verbose_name="Zone ID")), ("zone_id", models.CharField(max_length=255, verbose_name="Zone ID")),
("stop_url", models.URLField(blank=True, verbose_name="Stop URL")), ("url", models.URLField(blank=True, verbose_name="Stop URL")),
( (
"location_type", "location_type",
models.IntegerField( models.IntegerField(
@ -252,7 +298,7 @@ class Migration(migrations.Migration):
), ),
), ),
( (
"stop_timezone", "timezone",
models.CharField( models.CharField(
blank=True, max_length=255, verbose_name="Stop timezone" blank=True, max_length=255, verbose_name="Stop timezone"
), ),
@ -278,13 +324,17 @@ class Migration(migrations.Migration):
), ),
( (
"platform_code", "platform_code",
models.CharField(max_length=255, verbose_name="Platform code"), models.CharField(
blank=True, max_length=255, verbose_name="Platform code"
),
), ),
( (
"parent_station", "parent_station",
models.ForeignKey( models.ForeignKey(
blank=True, blank=True,
null=True,
on_delete=django.db.models.deletion.PROTECT, on_delete=django.db.models.deletion.PROTECT,
related_name="children",
to="sncfgtfs.stop", to="sncfgtfs.stop",
verbose_name="Parent station", verbose_name="Parent station",
), ),
@ -300,8 +350,8 @@ class Migration(migrations.Migration):
fields=[ fields=[
( (
"id", "id",
models.BigAutoField( models.CharField(
auto_created=True, max_length=255,
primary_key=True, primary_key=True,
serialize=False, serialize=False,
verbose_name="ID", verbose_name="ID",
@ -354,7 +404,7 @@ class Migration(migrations.Migration):
name="Trip", name="Trip",
fields=[ fields=[
( (
"trip_id", "id",
models.CharField( models.CharField(
max_length=255, max_length=255,
primary_key=True, primary_key=True,
@ -363,17 +413,13 @@ class Migration(migrations.Migration):
), ),
), ),
( (
"service_id", "headsign",
models.CharField(max_length=255, verbose_name="Service ID"),
),
(
"trip_headsign",
models.CharField( models.CharField(
blank=True, max_length=255, verbose_name="Trip headsign" blank=True, max_length=255, verbose_name="Trip headsign"
), ),
), ),
( (
"trip_short_name", "short_name",
models.CharField( models.CharField(
blank=True, max_length=255, verbose_name="Trip short name" blank=True, max_length=255, verbose_name="Trip short name"
), ),
@ -381,8 +427,8 @@ class Migration(migrations.Migration):
( (
"direction_id", "direction_id",
models.IntegerField( models.IntegerField(
blank=True,
choices=[(0, "Outbound"), (1, "Inbound")], choices=[(0, "Outbound"), (1, "Inbound")],
null=True,
verbose_name="Direction", verbose_name="Direction",
), ),
), ),
@ -401,13 +447,13 @@ class Migration(migrations.Migration):
( (
"wheelchair_accessible", "wheelchair_accessible",
models.IntegerField( models.IntegerField(
blank=True,
choices=[ choices=[
(0, "No information"), (0, "No information"),
(1, "Possible"), (1, "Possible"),
(2, "Not possible"), (2, "Not possible"),
], ],
default=0, default=0,
null=True,
verbose_name="Wheelchair accessible", verbose_name="Wheelchair accessible",
), ),
), ),
@ -420,6 +466,7 @@ class Migration(migrations.Migration):
(2, "Not possible"), (2, "Not possible"),
], ],
default=0, default=0,
null=True,
verbose_name="Bikes allowed", verbose_name="Bikes allowed",
), ),
), ),
@ -427,8 +474,18 @@ class Migration(migrations.Migration):
"route", "route",
models.ForeignKey( models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE, on_delete=django.db.models.deletion.CASCADE,
related_name="trips",
to="sncfgtfs.route", to="sncfgtfs.route",
verbose_name="Route ID", verbose_name="Route",
),
),
(
"service",
models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE,
related_name="trips",
to="sncfgtfs.calendar",
verbose_name="Service",
), ),
), ),
], ],
@ -442,8 +499,8 @@ class Migration(migrations.Migration):
fields=[ fields=[
( (
"id", "id",
models.BigAutoField( models.CharField(
auto_created=True, max_length=255,
primary_key=True, primary_key=True,
serialize=False, serialize=False,
verbose_name="ID", verbose_name="ID",
@ -451,6 +508,10 @@ class Migration(migrations.Migration):
), ),
("arrival_time", models.TimeField(verbose_name="Arrival time")), ("arrival_time", models.TimeField(verbose_name="Arrival time")),
("departure_time", models.TimeField(verbose_name="Departure time")), ("departure_time", models.TimeField(verbose_name="Departure time")),
(
"arrival_next_day",
models.BooleanField(default=False, verbose_name="Arrival next day"),
),
("stop_sequence", models.IntegerField(verbose_name="Stop sequence")), ("stop_sequence", models.IntegerField(verbose_name="Stop sequence")),
( (
"stop_headsign", "stop_headsign",
@ -461,7 +522,6 @@ class Migration(migrations.Migration):
( (
"pickup_type", "pickup_type",
models.IntegerField( models.IntegerField(
blank=True,
choices=[ choices=[
(0, "Regular"), (0, "Regular"),
(1, "None"), (1, "None"),
@ -469,13 +529,13 @@ class Migration(migrations.Migration):
(3, "Must coordinate with driver"), (3, "Must coordinate with driver"),
], ],
default=0, default=0,
null=True,
verbose_name="Pickup type", verbose_name="Pickup type",
), ),
), ),
( (
"drop_off_type", "drop_off_type",
models.IntegerField( models.IntegerField(
blank=True,
choices=[ choices=[
(0, "Regular"), (0, "Regular"),
(1, "None"), (1, "None"),
@ -483,29 +543,32 @@ class Migration(migrations.Migration):
(3, "Must coordinate with driver"), (3, "Must coordinate with driver"),
], ],
default=0, default=0,
null=True,
verbose_name="Drop off type", verbose_name="Drop off type",
), ),
), ),
( (
"timepoint", "timepoint",
models.BooleanField( models.BooleanField(
blank=True, default=True, verbose_name="Timepoint" default=True, null=True, verbose_name="Timepoint"
), ),
), ),
( (
"stop", "stop",
models.ForeignKey( models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE, on_delete=django.db.models.deletion.CASCADE,
related_name="stop_times",
to="sncfgtfs.stop", to="sncfgtfs.stop",
verbose_name="Stop ID", verbose_name="Stop ID",
), ),
), ),
( (
"trip_id", "trip",
models.ForeignKey( models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE, on_delete=django.db.models.deletion.CASCADE,
related_name="stop_times",
to="sncfgtfs.trip", to="sncfgtfs.trip",
verbose_name="Trip ID", verbose_name="Trip",
), ),
), ),
], ],

178
sncfgtfs/models.py
View File

@ -2,6 +2,13 @@ from django.db import models
from django.utils.translation import gettext_lazy as _ from django.utils.translation import gettext_lazy as _
class TransportType(models.TextChoices):
TGV = "TGV", _("TGV")
TER = "TER", _("TER")
INTERCITES = "IC", _("Intercités")
TRANSILIEN = "TN", _("Transilien")
class LocationType(models.IntegerChoices): class LocationType(models.IntegerChoices):
STOP_PLATFORM = 0, _("Stop/platform") STOP_PLATFORM = 0, _("Stop/platform")
STATION = 1, _("Station") STATION = 1, _("Station")
@ -46,68 +53,87 @@ class TransferType(models.IntegerChoices):
NOT_POSSIBLE = 3, _("Not possible") NOT_POSSIBLE = 3, _("Not possible")
class ExceptionType(models.IntegerChoices):
ADDED = 1, _("Added")
REMOVED = 2, _("Removed")
class Agency(models.Model): class Agency(models.Model):
agency_id = models.CharField( id = models.CharField(
max_length=255, max_length=255,
primary_key=True, primary_key=True,
verbose_name=_("Agency ID"), verbose_name=_("Agency ID"),
) )
agency_name = models.CharField( name = models.CharField(
max_length=255, max_length=255,
unique=True, unique=True,
verbose_name=_("Agency name"), verbose_name=_("Agency name"),
) )
agency_url = models.URLField( url = models.URLField(
verbose_name=_("Agency URL"), verbose_name=_("Agency URL"),
) )
agency_timezone = models.CharField( timezone = models.CharField(
max_length=255, max_length=255,
verbose_name=_("Agency timezone"), verbose_name=_("Agency timezone"),
) )
agency_lang = models.CharField( lang = models.CharField(
max_length=255, max_length=255,
verbose_name=_("Agency language"), verbose_name=_("Agency language"),
blank=True, blank=True,
) )
phone = models.CharField(
max_length=255,
verbose_name=_("Agency phone"),
blank=True,
)
email = models.EmailField(
verbose_name=_("Agency email"),
blank=True,
)
def __str__(self):
return self.name
class Meta: class Meta:
verbose_name = _("Agency") verbose_name = _("Agency")
verbose_name_plural = _("Agencies") verbose_name_plural = _("Agencies")
class Stop(models.Model): class Stop(models.Model):
stop_id = models.CharField( id = models.CharField(
max_length=255, max_length=255,
primary_key=True, primary_key=True,
verbose_name=_("Stop ID"), verbose_name=_("Stop ID"),
) )
stop_code = models.CharField( code = models.CharField(
max_length=255, max_length=255,
verbose_name=_("Stop code"), verbose_name=_("Stop code"),
blank=True, blank=True,
) )
stop_name = models.CharField( name = models.CharField(
max_length=255, max_length=255,
verbose_name=_("Stop name"), verbose_name=_("Stop name"),
) )
stop_desc = models.CharField( desc = models.CharField(
max_length=255, max_length=255,
verbose_name=_("Stop description"), verbose_name=_("Stop description"),
blank=True, blank=True,
) )
stop_lon = models.FloatField( lon = models.FloatField(
verbose_name=_("Stop longitude"), verbose_name=_("Stop longitude"),
) )
stop_lat = models.FloatField( lat = models.FloatField(
verbose_name=_("Stop latitude"), verbose_name=_("Stop latitude"),
) )
@ -116,7 +142,7 @@ class Stop(models.Model):
verbose_name=_("Zone ID"), verbose_name=_("Zone ID"),
) )
stop_url = models.URLField( url = models.URLField(
verbose_name=_("Stop URL"), verbose_name=_("Stop URL"),
blank=True, blank=True,
) )
@ -132,10 +158,12 @@ class Stop(models.Model):
to="Stop", to="Stop",
on_delete=models.PROTECT, on_delete=models.PROTECT,
verbose_name=_("Parent station"), verbose_name=_("Parent station"),
related_name="children",
blank=True, blank=True,
null=True,
) )
stop_timezone = models.CharField( timezone = models.CharField(
max_length=255, max_length=255,
verbose_name=_("Stop timezone"), verbose_name=_("Stop timezone"),
blank=True, blank=True,
@ -157,71 +185,79 @@ class Stop(models.Model):
platform_code = models.CharField( platform_code = models.CharField(
max_length=255, max_length=255,
verbose_name=_("Platform code"), verbose_name=_("Platform code"),
blank=True,
) )
def __str__(self):
return f"{self.name} ({self.id})"
class Meta: class Meta:
verbose_name = _("Stop") verbose_name = _("Stop")
verbose_name_plural = _("Stops") verbose_name_plural = _("Stops")
class Route(models.Model): class Route(models.Model):
route_id = models.CharField( id = models.CharField(
max_length=255, max_length=255,
primary_key=True, primary_key=True,
verbose_name=_("Route ID"), verbose_name=_("ID"),
) )
agency = models.ForeignKey( agency = models.ForeignKey(
to="Agency", to="Agency",
on_delete=models.CASCADE, on_delete=models.CASCADE,
verbose_name=_("Agency ID"), verbose_name=_("Agency ID"),
related_name="routes",
) )
route_short_name = models.CharField( short_name = models.CharField(
max_length=255, max_length=255,
verbose_name=_("Route short name"), verbose_name=_("Route short name"),
) )
route_long_name = models.CharField( long_name = models.CharField(
max_length=255, max_length=255,
verbose_name=_("Route long name"), verbose_name=_("Route long name"),
) )
route_desc = models.CharField( desc = models.CharField(
max_length=255, max_length=255,
verbose_name=_("Route description"), verbose_name=_("Route description"),
blank=True, blank=True,
) )
route_type = models.IntegerField( type = models.IntegerField(
verbose_name=_("Route type"), verbose_name=_("Route type"),
choices=RouteType, choices=RouteType,
) )
route_url = models.URLField( url = models.URLField(
verbose_name=_("Route URL"), verbose_name=_("Route URL"),
blank=True, blank=True,
) )
route_color = models.CharField( color = models.CharField(
max_length=255, max_length=255,
verbose_name=_("Route color"), verbose_name=_("Route color"),
blank=True, blank=True,
) )
route_text_color = models.CharField( text_color = models.CharField(
max_length=255, max_length=255,
verbose_name=_("Route text color"), verbose_name=_("Route text color"),
blank=True, blank=True,
) )
def __str__(self):
return f"{self.long_name}"
class Meta: class Meta:
verbose_name = _("Route") verbose_name = _("Route")
verbose_name_plural = _("Routes") verbose_name_plural = _("Routes")
class Trip(models.Model): class Trip(models.Model):
trip_id = models.CharField( id = models.CharField(
max_length=255, max_length=255,
primary_key=True, primary_key=True,
verbose_name=_("Trip ID"), verbose_name=_("Trip ID"),
@ -230,21 +266,24 @@ class Trip(models.Model):
route = models.ForeignKey( route = models.ForeignKey(
to="Route", to="Route",
on_delete=models.CASCADE, on_delete=models.CASCADE,
verbose_name=_("Route ID"), verbose_name=_("Route"),
related_name="trips",
) )
service_id = models.CharField( service = models.ForeignKey(
max_length=255, to="Calendar",
verbose_name=_("Service ID"), on_delete=models.CASCADE,
verbose_name=_("Service"),
related_name="trips",
) )
trip_headsign = models.CharField( headsign = models.CharField(
max_length=255, max_length=255,
verbose_name=_("Trip headsign"), verbose_name=_("Trip headsign"),
blank=True, blank=True,
) )
trip_short_name = models.CharField( short_name = models.CharField(
max_length=255, max_length=255,
verbose_name=_("Trip short name"), verbose_name=_("Trip short name"),
blank=True, blank=True,
@ -253,7 +292,7 @@ class Trip(models.Model):
direction_id = models.IntegerField( direction_id = models.IntegerField(
verbose_name=_("Direction"), verbose_name=_("Direction"),
choices=Direction, choices=Direction,
blank=True, null=True,
) )
block_id = models.CharField( block_id = models.CharField(
@ -272,25 +311,40 @@ class Trip(models.Model):
verbose_name=_("Wheelchair accessible"), verbose_name=_("Wheelchair accessible"),
choices=AccessInformation, choices=AccessInformation,
default=AccessInformation.NO_INFORMATION, default=AccessInformation.NO_INFORMATION,
blank=True, null=True,
) )
bikes_allowed = models.IntegerField( bikes_allowed = models.IntegerField(
verbose_name=_("Bikes allowed"), verbose_name=_("Bikes allowed"),
choices=AccessInformation, choices=AccessInformation,
default=AccessInformation.NO_INFORMATION, default=AccessInformation.NO_INFORMATION,
null=True,
) )
@property
def destination(self):
return self.stop_times.order_by('-stop_sequence').first().stop.name
def __str__(self):
return f"{self.route.long_name} - {self.id}"
class Meta: class Meta:
verbose_name = _("Trip") verbose_name = _("Trip")
verbose_name_plural = _("Trips") verbose_name_plural = _("Trips")
class StopTime(models.Model): class StopTime(models.Model):
trip_id = models.ForeignKey( id = models.CharField(
max_length=255,
primary_key=True,
verbose_name=_("ID"),
)
trip = models.ForeignKey(
to="Trip", to="Trip",
on_delete=models.CASCADE, on_delete=models.CASCADE,
verbose_name=_("Trip ID"), verbose_name=_("Trip"),
related_name="stop_times",
) )
arrival_time = models.TimeField( arrival_time = models.TimeField(
@ -301,10 +355,16 @@ class StopTime(models.Model):
verbose_name=_("Departure time"), verbose_name=_("Departure time"),
) )
arrival_next_day = models.BooleanField(
verbose_name=_("Arrival next day"),
default=False,
)
stop = models.ForeignKey( stop = models.ForeignKey(
to="Stop", to="Stop",
on_delete=models.CASCADE, on_delete=models.CASCADE,
verbose_name=_("Stop ID"), verbose_name=_("Stop ID"),
related_name="stop_times",
) )
stop_sequence = models.IntegerField( stop_sequence = models.IntegerField(
@ -321,29 +381,32 @@ class StopTime(models.Model):
verbose_name=_("Pickup type"), verbose_name=_("Pickup type"),
choices=PickupType, choices=PickupType,
default=PickupType.REGULAR, default=PickupType.REGULAR,
blank=True, null=True,
) )
drop_off_type = models.IntegerField( drop_off_type = models.IntegerField(
verbose_name=_("Drop off type"), verbose_name=_("Drop off type"),
choices=PickupType, choices=PickupType,
default=PickupType.REGULAR, default=PickupType.REGULAR,
blank=True, null=True,
) )
timepoint = models.BooleanField( timepoint = models.BooleanField(
verbose_name=_("Timepoint"), verbose_name=_("Timepoint"),
default=True, default=True,
blank=True, null=True,
) )
def __str__(self):
return f"{self.trip.route.long_name} - {self.trip_id} - {self.stop.name}"
class Meta: class Meta:
verbose_name = _("Stop time") verbose_name = _("Stop time")
verbose_name_plural = _("Stop times") verbose_name_plural = _("Stop times")
class Calendar(models.Model): class Calendar(models.Model):
service_id = models.CharField( id = models.CharField(
max_length=255, max_length=255,
primary_key=True, primary_key=True,
verbose_name=_("Service ID"), verbose_name=_("Service ID"),
@ -385,16 +448,32 @@ class Calendar(models.Model):
verbose_name=_("End date"), verbose_name=_("End date"),
) )
transport_type = models.CharField(
max_length=255,
verbose_name=_("Transport type"),
choices=TransportType,
)
def __str__(self):
return self.id
class Meta: class Meta:
verbose_name = _("Calendar") verbose_name = _("Calendar")
verbose_name_plural = _("Calendars") verbose_name_plural = _("Calendars")
class CalendarDate(models.Model): class CalendarDate(models.Model):
service_id = models.ForeignKey( id = models.CharField(
max_length=255,
primary_key=True,
verbose_name=_("ID"),
)
service = models.ForeignKey(
to="Calendar", to="Calendar",
on_delete=models.CASCADE, on_delete=models.CASCADE,
verbose_name=_("Service ID"), verbose_name=_("Service"),
related_name="dates",
) )
date = models.DateField( date = models.DateField(
@ -403,14 +482,30 @@ class CalendarDate(models.Model):
exception_type = models.IntegerField( exception_type = models.IntegerField(
verbose_name=_("Exception type"), verbose_name=_("Exception type"),
choices=ExceptionType,
) )
transport_type = models.CharField(
max_length=255,
verbose_name=_("Transport type"),
choices=TransportType,
)
def __str__(self):
return f"{self.service.id} - {self.date} - {self.exception_type}"
class Meta: class Meta:
verbose_name = _("Calendar date") verbose_name = _("Calendar date")
verbose_name_plural = _("Calendar dates") verbose_name_plural = _("Calendar dates")
class Transfer(models.Model): class Transfer(models.Model):
id = models.CharField(
max_length=255,
primary_key=True,
verbose_name=_("ID"),
)
from_stop = models.ForeignKey( from_stop = models.ForeignKey(
to="Stop", to="Stop",
on_delete=models.CASCADE, on_delete=models.CASCADE,
@ -442,11 +537,6 @@ class Transfer(models.Model):
class FeedInfo(models.Model): class FeedInfo(models.Model):
feed_id = models.SmallIntegerField(
primary_key=True,
verbose_name=_("Feed ID"),
)
feed_publisher_name = models.CharField( feed_publisher_name = models.CharField(
max_length=255, max_length=255,
verbose_name=_("Feed publisher name"), verbose_name=_("Feed publisher name"),