-
Notifications
You must be signed in to change notification settings - Fork 2
/
import_stations.py
157 lines (137 loc) · 8.19 KB
/
import_stations.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
#!/usr/bin/env python
from __future__ import annotations
import argparse
import logging
import os
import re
from os import PathLike
from typing import List, Tuple, Optional, Any, Dict
import gpxpy.gpx
from cli_utils import check_files, process_station_input, add_default_cli_args, add_station_cli_args, \
parse_station_args, use_default_cli_args
from geo.location_data import add_location_data_to_list, with_location_data
from structures import DataSet
from structures.country import split_country, CountryRepresentation, iso_3166_to_country, tld_to_country
from structures.station import iter_stations_by_codes_reverse, Station
from tc_utils import TcFile
from tc_utils.stations import add_stations_to_file
def import_stations_into_tc(station_codes: List[str],
tc_directory: PathLike | str = '..',
data_directory: PathLike | str = 'data',
override_stations: bool = False,
update_stations: bool = False,
append: bool = False,
trassenfinder: bool = False,
gpx: bool = False,
annotate: bool = False,
data_set: Optional[DataSet] = None,
station_json: TcFile | None = None,
case_sensitive: bool = False,
use_google: bool = False,
**kwargs) -> TcFile:
if not data_set:
data_set = DataSet.load_data(data_directory)
code_to_station = {code: station for code, station in iter_stations_by_codes_reverse(data_set.station_data)}
stations = [code_to_station[code.upper() if not case_sensitive else code] for code in
station_codes]
add_location_data_to_list(stations, use_google=use_google)
for station in stations:
if not station.platform_count or not station.platform_length and station.group != 4:
logging.warning("{} : Keine Bahnsteigdaten bekannt".format(station.codes[0]))
if station.location:
logging.warning("{} auf OSM: https://openstreetmap.org/#map=17/{}/{}&layers=T"
.format(station.codes[0],
station.location.latitude,
station.location.longitude))
logging.warning("{} auf G/M: https://maps.google.com/maps/@{},{},17z/data=!3m1!1e3"
.format(station.codes[0],
station.location.latitude,
station.location.longitude))
if annotate:
station['google_maps'] = "https://maps.google.com/maps/@{},{},17z/data=!3m1!1e3".format(
station.location.latitude, station.location.longitude)
station['osm'] = "https://openstreetmap.org/#map=17/{}/{}&layers=T".format(
station.location.latitude, station.location.longitude)
else:
"{} ohne bekannten Standort".format(station.codes[0])
if not station_json:
station_json = TcFile('Station', tc_directory)
add_stations_to_file(stations.copy(), station_json, override_stations, update_stations, append=append)
if trassenfinder:
create_trassenfinder(
[(code, code_to_station[code.upper() if not case_sensitive else code].name) for code in station_codes],
tc_directory)
if gpx:
create_gpx(stations, tc_directory)
return station_json
def get_filename(code_start: str, code_end: str, extension: str, tc_directory: PathLike | str = '..'):
country_start, code_start, representation = split_country(code_start)
if country_start is not None and representation in (CountryRepresentation.FLAG,):
code_start = country_start.iso_3166 + "_" + code_start
country_end, code_end, _ = split_country(code_end)
if country_end is not None and representation in (CountryRepresentation.FLAG,):
code_end = country_end.iso_3166 + "_" + code_end
filename = "{}-{}.{}".format(code_start, code_end, extension)
index = 1
while os.path.exists(os.path.join(tc_directory, filename)):
filename = "{}-{}-{}.{}".format(code_start, code_end, index, extension)
index += 1
return os.path.join(tc_directory, filename)
def create_trassenfinder(stations: List[Tuple[str, str]], tc_directory: PathLike | str = '..'):
lines = ['utf-8;"Lfd. km";;"Betriebsstelle (kurz)";"Nachfolgende Streckennr.";;;;;;;;;;;;;;;"Bemerkung"\n']
for index, (code, station_name) in enumerate(stations):
lines.append('"{}";;"{}";"0";;;;;;;;;;;;;;"{}; Kundenhalt"\n'.format("0,0" if index == 0 else "", code, station_name))
assert len(lines) == len(stations) + 1
filename = get_filename(stations[0][0], stations[-1][0], "csv", tc_directory)
with open(filename, 'w', encoding='utf-8', newline='\n') as outfile:
outfile.writelines(lines)
def create_gpx(stations: List[Station], tc_directory: PathLike | str = '..'):
gpx = gpxpy.gpx.GPX()
for station in stations:
if not station.location:
logging.warning("Haltestelle ohne Standortdaten: {}. Bitte manuell hinzufügen.".format(station.codes[0]))
trackpoints = [gpxpy.gpx.GPXTrackPoint(latitude=station.location.latitude,
longitude=station.location.longitude,
name=station.name) for station in stations if station.location]
track = gpxpy.gpx.GPXTrack()
segment = gpxpy.gpx.GPXTrackSegment()
segment.points.extend(trackpoints)
track.segments.append(segment)
gpx.tracks.append(track)
filename = get_filename(stations[0].codes[0], stations[-1].codes[0], "gpx", tc_directory)
with open(filename, 'w', encoding='utf-8', newline='\n') as outfile:
outfile.write(gpx.to_xml(prettyprint=True))
if __name__ == '__main__':
parser = argparse.ArgumentParser(description='Importiere neue Betriebsstellen in TrainCompany')
add_default_cli_args(parser)
add_station_cli_args(parser,
help="Die (RIL100-)Codes der Haltestellen, die hinzugefügt werden sollen",
help_countries="Die Länder, deren Haltestellen hinzugefügt werden sollen",
required=True,
# TODO: Allow this
allow_multiple_stations=True)
update_or_override = parser.add_mutually_exclusive_group()
update_or_override.add_argument('--override-stations', action='store_true',
help="Überschreibt Haltestellen, bzw. fügt spezifischere hinzu")
update_or_override.add_argument('--update-stations', action='store_true',
help="Aktualisiert existierende Haltestellen, fügt aber keine hinzu")
parser.add_argument('--append', action='store_true',
help="Fügt alles am Ende ein")
parser.add_argument('--annotate', action='store_true',
help="Fügt Links zu Google Maps/OSM hinzu, wenn Bahnsteigdaten fehlen")
parser.add_argument('--trassenfinder', action='store_true',
help="Legt eine Trassenfinder-ähnliche Datei mit den Haltestellen an.")
parser.add_argument('--gpx', action='store_true',
help="Legt eine GPX-Datei mit Wegpunkten an, die bspw. auf brouter.de importiert werden kann.")
parser.add_argument("--use-google", action='store_true',
help="(VERALTET) Nutzt die Google Maps-API statt Komoot Photon für fehlende Standort-Daten (API-Key erforderlich)")
args = parser.parse_args()
use_default_cli_args(args)
check_files(args.tc_directory, args.data_directory)
stations = parse_station_args(args, required=True, inplace=False)
station_json = TcFile("Station", args.tc_directory)
data_set = DataSet.load_data(args.data_directory)
for station_codes in stations:
station_json = import_stations_into_tc(**args.__dict__, data_set=data_set, station_json=station_json,
station_codes=station_codes)
station_json.save()