return result
-def update_sledrun(vao: Vao, site: WikiSite, title: str, query_date: date):
+def update_sledrun(vao: Vao, site: WikiSite, title: str, query_date: date, missing_only: bool, no_schedules: bool):
sledrun_json_page = site.query_page(f'{title}/Rodelbahn.json')
sledrun_json = page_json(sledrun_json_page)
sledrun_json_orig = sledrun_json.copy()
- pos = sledrun_json['bottom']['position']
+ if missing_only:
+ if pt_stops := sledrun_json.get('public_transport_stops'):
+ if len(pt_stops) > 0 and pt_stops[0].get('ifopt_stop_id'):
+ return
+
+ pos = sledrun_json.get('bottom', {}).get('position')
+ if not pos:
+ return
lon_lat = LonLat(pos['longitude'], pos['latitude'])
nearby_stops, line_info_dict = try_vao_nearby_stops(vao, lon_lat)
nearby_stops = unique_nearby_stops(nearby_stops)
journey_minutes = 1439
public_transport_stops = []
+ lines = []
for stop_with_dist in stops_with_dists:
- departures = try_vao_departure_board(vao, stop_with_dist.stop.ext_id, journey_date, journey_minutes,
- line_info_dict)
- arrivals = try_vao_arrival_board(vao, stop_with_dist.stop.ext_id, journey_date, journey_minutes, line_info_dict)
- # journey_detail_ref = arrivals[41].detail_ref
- # journey_detail = try_vao_journey_detail(vao, journey_detail_ref)
-
- vao_line_ids = set(a.vao_line_id for a in arrivals).union(d.vao_line_id for d in departures) \
- .union(stop_with_dist.stop.line_ids)
- lines = []
-
- for vao_line_id in vao_line_ids:
- line_info = line_info_dict[vao_line_id]
- departure = []
- for direction in set(d.direction for d in departures if d.vao_line_id == vao_line_id):
- departure.append({
- "direction": direction,
- "datetime": [d.time.isoformat(timespec='minutes') for d in departures
- if d.vao_line_id == vao_line_id and d.direction == direction]
- })
-
- arrival = []
- for origin in set(a.origin for a in arrivals if a.vao_line_id == vao_line_id):
- arrival.append({
- "origin": origin,
- "datetime": [a.time.isoformat(timespec='minutes') for a in arrivals
- if a.vao_line_id == vao_line_id and a.origin == origin]
- })
-
- schedules = [{
- 'service_date': journey_date.date().isoformat(),
- "day_type": "work_day",
- "departure": departure,
- "arrival": arrival,
- }]
-
- line = {
- "vao_line_id": vao_line_id,
- "line": line_info.line,
- "category": line_info.category,
- "operator": line_info.operator,
- "schedules": schedules,
- }
- lines.append(line)
+ if not no_schedules:
+ departures = try_vao_departure_board(vao, stop_with_dist.stop.ext_id, journey_date, journey_minutes,
+ line_info_dict)
+ arrivals = try_vao_arrival_board(vao, stop_with_dist.stop.ext_id, journey_date, journey_minutes, line_info_dict)
+ # journey_detail_ref = arrivals[41].detail_ref
+ # journey_detail = try_vao_journey_detail(vao, journey_detail_ref)
+
+ vao_line_ids = set(a.vao_line_id for a in arrivals).union(d.vao_line_id for d in departures) \
+ .union(stop_with_dist.stop.line_ids)
+ lines = []
+
+ for vao_line_id in vao_line_ids:
+ line_info = line_info_dict[vao_line_id]
+ departure = []
+ for direction in set(d.direction for d in departures if d.vao_line_id == vao_line_id):
+ departure.append({
+ "direction": direction,
+ "datetime": [d.time.isoformat(timespec='minutes') for d in departures
+ if d.vao_line_id == vao_line_id and d.direction == direction]
+ })
+
+ arrival = []
+ for origin in set(a.origin for a in arrivals if a.vao_line_id == vao_line_id):
+ arrival.append({
+ "origin": origin,
+ "datetime": [a.time.isoformat(timespec='minutes') for a in arrivals
+ if a.vao_line_id == vao_line_id and a.origin == origin]
+ })
+
+ schedules = [{
+ 'service_date': journey_date.date().isoformat(),
+ "day_type": "work_day",
+ "departure": departure,
+ "arrival": arrival,
+ }]
+
+ line = {
+ "vao_line_id": vao_line_id,
+ "line": line_info.line,
+ "category": line_info.category,
+ "operator": line_info.operator,
+ "schedules": schedules,
+ }
+ lines.append(line)
public_transport_stop = {
"name": stop_with_dist.stop.name,
},
"walk_distance": stop_with_dist.dist.dist_m,
"walk_time": stop_with_dist.dist.time_min,
- "lines": lines,
}
+ if not no_schedules:
+ public_transport_stop["lines"] = lines
if ifopt_stop_id := vao_ext_id_to_ifopt_stop_id(stop_with_dist.stop.ext_id):
public_transport_stop['ifopt_stop_id'] = ifopt_stop_id
vvt_stop_id = get_vvt_stop_id(stop_with_dist.stop.ext_id)
public_transport_stops.append(public_transport_stop)
- sledrun_json['public_transport_stops'] = public_transport_stops
+ if len(public_transport_stops) > 0 or 'public_transport_stops' not in sledrun_json:
+ sledrun_json['public_transport_stops'] = public_transport_stops
if sledrun_json == sledrun_json_orig:
return
)
-def update_public_transport(ini_files: List[str], query_date: date, sledrun: Optional[str]):
+def update_public_transport(ini_files: List[str], query_date: date, sledrun: Optional[str], missing_only: bool,
+ no_schedules: bool):
config = configparser.ConfigParser()
config.read(ini_files)
vao = Vao(config.get('vao', 'access_id'))
if sledrun is not None:
- update_sledrun(vao, site, sledrun, query_date)
+ update_sledrun(vao, site, sledrun, query_date, missing_only, no_schedules)
return
for result in site.query(list='categorymembers', cmtitle='Kategorie:Rodelbahn', cmlimit='max'):
for page in result['categorymembers']:
sledrun = page['title']
print(sledrun)
- update_sledrun(vao, site, sledrun, query_date)
+ update_sledrun(vao, site, sledrun, query_date, missing_only, no_schedules)
def main():
query_date = default_query_date(date.today())
parser = argparse.ArgumentParser(description='Update public transport information in sledrun JSON files.')
parser.add_argument('--sledrun', help='If given, work on a single sled run page, otherwise at the whole category.')
+ parser.add_argument('--missing', action='store_true',
+ help='If given, only work on stops that have no ifopt_stop_id.')
+ parser.add_argument('--no-schedules', action='store_true', help='If given, don\'t add schedules.')
parser.add_argument('--date', type=date.fromisoformat, default=query_date,
help='Working week date to query the database.')
parser.add_argument('inifile', nargs='+', help='inifile.ini, see: https://www.winterrodeln.org/trac/wiki/ConfigIni')
args = parser.parse_args()
- update_public_transport(args.inifile, args.date, args.sledrun)
+ update_public_transport(args.inifile, args.date, args.sledrun, args.missing, args.no_schedules)
if __name__ == '__main__':
ToastFreeware / philipp / winterrodeln / wrpylib.git / blobdiff